From 8a9fe633832d243bb29250d4d6c08bf662fa0fff Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Thu, 5 Nov 2026 13:02:42 -0800 Subject: [PATCH 1/3] Extract the live server sync harness into helpers The second server, second_environment and the sync session setup move from test_syncsession.py into integration/helpers.py, with the setup in a new BasePushPullClientTestCase. PushPullClientTestCase extends it and keeps its tests and the activity timestamp checks they use. Other test modules can now reuse the harness without importing PushPullClientTestCase, which would collect and run its tests again. Refs #363 Co-Authored-By: Claude Opus 5.5 --- tests/testapp/tests/integration/helpers.py | 192 ++++++++++++++++++ .../tests/integration/test_syncsession.py | 189 +---------------- 2 files changed, 197 insertions(+), 184 deletions(-) create mode 100644 tests/testapp/tests/integration/helpers.py diff --git a/tests/testapp/tests/integration/helpers.py b/tests/testapp/tests/integration/helpers.py new file mode 100644 index 00000000..62611886 --- /dev/null +++ b/tests/testapp/tests/integration/helpers.py @@ -0,0 +1,192 @@ +import contextlib +import json +import os +import subprocess +import sys +import time + +import pytest +import requests +from django.conf import settings +from django.test.testcases import TransactionTestCase +from facility_profile.models import MyUser +from requests.exceptions import RequestException +from testapp.settings import BASE_DIR + +from morango.models.certificates import Certificate +from morango.models.certificates import Filter +from morango.models.certificates import Key +from morango.models.certificates import ScopeDefinition +from morango.models.core import InstanceIDModel +from morango.sync.controller import MorangoProfileController + +from ..compat import EnvironmentVarGuard +from ..helpers import get_free_tcp_port + +SECOND_TEST_DATABASE = "default2" +SECOND_SYSTEM_ID = "default2" + + +@contextlib.contextmanager +def second_environment(): + with EnvironmentVarGuard() as env: + env["MORANGO_TEST_DATABASE"] = SECOND_TEST_DATABASE + env["MORANGO_SYSTEM_ID"] = SECOND_SYSTEM_ID + instance2, _ = InstanceIDModel.get_or_create_current_instance(clear_cache=True) + yield + instance1, _ = InstanceIDModel.get_or_create_current_instance(clear_cache=True) + assert instance1.id != instance2.id + + +class LiveServer: + def __init__(self): + self.env = os.environ.copy() + self.env["MORANGO_SYSTEM_ID"] = SECOND_SYSTEM_ID + self.port = get_free_tcp_port() + self.host = "127.0.0.1" + self.start() + + @property + def baseurl(self): + return f"http://{self.host}:{self.port}/" + + def start(self): + manage_py_path = os.path.join(BASE_DIR, "manage.py") + self._instance = subprocess.Popen( + [ + sys.executable, + manage_py_path, + "runserver", + "--nothreading", + "--noreload", + "--settings", + "testapp.server2_settings", + f"{self.host}:{self.port}", + ], + env=self.env, + ) + self._wait_for_server_start() + + def _wait_for_server_start(self, timeout=20): + for i in range(timeout * 2): + try: + resp = requests.get(self.baseurl, timeout=3) + if resp.status_code > 0: + return + except RequestException: + pass + time.sleep(0.5) + + raise Exception("Server did not start within {} seconds".format(timeout)) + + def kill(self): + try: + self._instance.kill() + except OSError: + pass + + +@pytest.mark.skipif(getattr(settings, "MORANGO_TEST_POSTGRESQL", False), reason="Not supported") +class BasePushPullClientTestCase(TransactionTestCase): + """ + Runs a second morango instance as a live server, and sets up a sync session with it + """ + + profile = "facilitydata" + databases = ["default", SECOND_TEST_DATABASE] + + @classmethod + def setUpClass(cls): + super(TransactionTestCase, cls).setUpClass() + cls.server = LiveServer() + + @classmethod + def tearDownClass(cls): + # There may not be a 'server' attribute if setUpClass() for some + # reasons has raised an exception. + if hasattr(cls, "server"): + # Terminate the live server's thread + cls.server.kill() + super(TransactionTestCase, cls).tearDownClass() + + def setUp(self): + super(BasePushPullClientTestCase, self).setUp() + self.profile_controller = MorangoProfileController(self.profile) + self.conn = self.profile_controller.create_network_connection(self.server.baseurl) + self.conn.chunk_size = 3 + + self.remote_user, self.root_cert_id = self._setUpServer() + self.filter = Filter("{}:user".format(self.remote_user.id)) + self.client = self._setUpClient(self.root_cert_id) + self.session = self.client.sync_session + + # perform an initial sync to ensure the user is on both sides + client = self.client.get_pull_client() + client.initialize(self.filter) + client.run() + client.finalize() + + self.local_user = MyUser.objects.first() + + def _setUpCertScopes(self): + root_scope = ScopeDefinition.objects.create( + id="root_scope", + profile=self.profile, + version=1, + primary_scope_param_key="user", + description="Root cert for ${user}.", + read_filter_template="", + write_filter_template="", + read_write_filter_template="${user}", + ) + + subset_scope = ScopeDefinition.objects.create( + id="subset_scope", + profile=self.profile, + version=1, + primary_scope_param_key="", + description="Subset cert under ${user} for :${sub}.", + read_filter_template="${user}", + write_filter_template="${user}:${sub}", + read_write_filter_template="", + ) + return root_scope, subset_scope + + def _setUpServer(self): + with second_environment(): + self.server_root_scope, self.server_subset_scope = self._setUpCertScopes() + root_cert = Certificate.generate_root_certificate(self.server_root_scope.id) + + remote_user = MyUser.objects.create( + _morango_source_id=root_cert.id, username="bob", is_superuser=True + ) + remote_user.set_password("password") + remote_user.save() + + subset_cert = Certificate( + parent=root_cert, + profile=self.profile, + scope_definition=self.server_subset_scope, + scope_version=self.server_subset_scope.version, + scope_params=json.dumps({"user": remote_user.id, "sub": "user"}), + private_key=Key(), + ) + root_cert.sign_certificate(subset_cert) + subset_cert.save() + return remote_user, root_cert.id + + def _setUpClient(self, primary_partition): + self.client_root_scope, self.client_subset_scope = self._setUpCertScopes() + + server_certs = self.conn.get_remote_certificates( + primary_partition, self.client_root_scope.id + ) + server_cert = server_certs[0] + client_cert = self.conn.certificate_signing_request( + server_cert, + self.client_subset_scope.id, + {"user": primary_partition, "sub": "user"}, + userargs="bob", + password="password", + ) + return self.conn.create_sync_session(client_cert, server_cert) diff --git a/tests/testapp/tests/integration/test_syncsession.py b/tests/testapp/tests/integration/test_syncsession.py index 40f33a37..f04b96d6 100644 --- a/tests/testapp/tests/integration/test_syncsession.py +++ b/tests/testapp/tests/integration/test_syncsession.py @@ -1,201 +1,22 @@ -import contextlib -import json -import os -import subprocess -import sys -import time - import mock -import pytest -import requests -from django.conf import settings -from django.test.testcases import TransactionTestCase from facility_profile.models import InteractionLog -from facility_profile.models import MyUser from facility_profile.models import SummaryLog -from requests.exceptions import RequestException from requests.exceptions import Timeout -from testapp.settings import BASE_DIR from morango.errors import MorangoError -from morango.models.certificates import Certificate -from morango.models.certificates import Filter -from morango.models.certificates import Key -from morango.models.certificates import ScopeDefinition from morango.models.core import Buffer -from morango.models.core import InstanceIDModel from morango.models.core import TransferSession -from morango.sync.controller import MorangoProfileController - -from ..compat import EnvironmentVarGuard -from ..helpers import get_free_tcp_port - -SECOND_TEST_DATABASE = "default2" -SECOND_SYSTEM_ID = "default2" - - -@contextlib.contextmanager -def second_environment(): - with EnvironmentVarGuard() as env: - env["MORANGO_TEST_DATABASE"] = SECOND_TEST_DATABASE - env["MORANGO_SYSTEM_ID"] = SECOND_SYSTEM_ID - instance2, _ = InstanceIDModel.get_or_create_current_instance(clear_cache=True) - yield - instance1, _ = InstanceIDModel.get_or_create_current_instance(clear_cache=True) - assert instance1.id != instance2.id - - -class LiveServer: - def __init__(self): - self.env = os.environ.copy() - self.env["MORANGO_SYSTEM_ID"] = SECOND_SYSTEM_ID - self.port = get_free_tcp_port() - self.host = "127.0.0.1" - self.start() - - @property - def baseurl(self): - return f"http://{self.host}:{self.port}/" - - def start(self): - manage_py_path = os.path.join(BASE_DIR, "manage.py") - self._instance = subprocess.Popen( - [ - sys.executable, - manage_py_path, - "runserver", - "--nothreading", - "--noreload", - "--settings", - "testapp.server2_settings", - f"{self.host}:{self.port}", - ], - env=self.env, - ) - self._wait_for_server_start() - - def _wait_for_server_start(self, timeout=20): - for i in range(timeout * 2): - try: - resp = requests.get(self.baseurl, timeout=3) - if resp.status_code > 0: - return - except RequestException: - pass - time.sleep(0.5) - - raise Exception("Server did not start within {} seconds".format(timeout)) - - def kill(self): - try: - self._instance.kill() - except OSError: - pass - - -@pytest.mark.skipif(getattr(settings, "MORANGO_TEST_POSTGRESQL", False), reason="Not supported") -class PushPullClientTestCase(TransactionTestCase): - profile = "facilitydata" - databases = ["default", SECOND_TEST_DATABASE] - - @classmethod - def setUpClass(cls): - super(TransactionTestCase, cls).setUpClass() - cls.server = LiveServer() - - @classmethod - def tearDownClass(cls): - # There may not be a 'server' attribute if setUpClass() for some - # reasons has raised an exception. - if hasattr(cls, "server"): - # Terminate the live server's thread - cls.server.kill() - super(TransactionTestCase, cls).tearDownClass() +from .helpers import BasePushPullClientTestCase +from .helpers import second_environment + + +class PushPullClientTestCase(BasePushPullClientTestCase): def setUp(self): super(PushPullClientTestCase, self).setUp() - self.profile_controller = MorangoProfileController(self.profile) - self.conn = self.profile_controller.create_network_connection(self.server.baseurl) - self.conn.chunk_size = 3 - - self.remote_user, self.root_cert_id = self._setUpServer() - self.filter = Filter("{}:user".format(self.remote_user.id)) - self.client = self._setUpClient(self.root_cert_id) - self.session = self.client.sync_session self.last_session_activity = self.session.last_activity_timestamp self.last_transfer_activity = None - # perform an initial sync to ensure the user is on both sides - client = self.client.get_pull_client() - client.initialize(self.filter) - client.run() - client.finalize() - - self.local_user = MyUser.objects.first() - - def _setUpCertScopes(self): - root_scope = ScopeDefinition.objects.create( - id="root_scope", - profile=self.profile, - version=1, - primary_scope_param_key="user", - description="Root cert for ${user}.", - read_filter_template="", - write_filter_template="", - read_write_filter_template="${user}", - ) - - subset_scope = ScopeDefinition.objects.create( - id="subset_scope", - profile=self.profile, - version=1, - primary_scope_param_key="", - description="Subset cert under ${user} for :${sub}.", - read_filter_template="${user}", - write_filter_template="${user}:${sub}", - read_write_filter_template="", - ) - return root_scope, subset_scope - - def _setUpServer(self): - with second_environment(): - self.server_root_scope, self.server_subset_scope = self._setUpCertScopes() - root_cert = Certificate.generate_root_certificate(self.server_root_scope.id) - - remote_user = MyUser.objects.create( - _morango_source_id=root_cert.id, username="bob", is_superuser=True - ) - remote_user.set_password("password") - remote_user.save() - - subset_cert = Certificate( - parent=root_cert, - profile=self.profile, - scope_definition=self.server_subset_scope, - scope_version=self.server_subset_scope.version, - scope_params=json.dumps({"user": remote_user.id, "sub": "user"}), - private_key=Key(), - ) - root_cert.sign_certificate(subset_cert) - subset_cert.save() - return remote_user, root_cert.id - - def _setUpClient(self, primary_partition): - self.client_root_scope, self.client_subset_scope = self._setUpCertScopes() - - server_certs = self.conn.get_remote_certificates( - primary_partition, self.client_root_scope.id - ) - server_cert = server_certs[0] - client_cert = self.conn.certificate_signing_request( - server_cert, - self.client_subset_scope.id, - {"user": primary_partition, "sub": "user"}, - userargs="bob", - password="password", - ) - return self.conn.create_sync_session(client_cert, server_cert) - def assertLastActivityUpdate(self, transfer_session=None): """A signal callable that asserts `last_activity_timestamp`s are updated""" if self.last_transfer_activity is not None: From 2b8a92b236d8c75d2af0f24c56797bbb0d750bc2 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Thu, 5 Nov 2026 13:02:42 -0800 Subject: [PATCH 2/3] Add characterization tests for async network sync Pins what a full push and a full pull with ASYNC_OPERATIONS send and store, so the refactors of the sync layer can show they change neither. A spy on SessionWrapper.request records each request, then passes the call through to the real method. For each request it records the method, path, query params, body encoding, body keys, the values of the body fields that are stable between runs, the record count of each pushed chunk, and the is_retryable flag. Generated IDs are normalized to . The tests compare the recording with fixed expected lists for every transfer stage. The tests also assert the final local transfer session, the Buffer, Store and app rows on both sides, and the byte counts the server stores. The spy fails a sync that sends more than 100 requests or runs longer than 120 seconds, so a stalled sync fails instead of hanging. Closes #363 Co-Authored-By: Claude Opus 5.5 --- .../test_network_characterization.py | 363 ++++++++++++++++++ 1 file changed, 363 insertions(+) create mode 100644 tests/testapp/tests/integration/test_network_characterization.py diff --git a/tests/testapp/tests/integration/test_network_characterization.py b/tests/testapp/tests/integration/test_network_characterization.py new file mode 100644 index 00000000..04dd326f --- /dev/null +++ b/tests/testapp/tests/integration/test_network_characterization.py @@ -0,0 +1,363 @@ +""" +Characterization tests that pin the HTTP requests and the final state of a network push and pull +that use `ASYNC_OPERATIONS`. Refactors of the sync layer must pass these tests without edits. +""" + +import gzip +import json +import re +import time + +import mock +from facility_profile.models import InteractionLog +from facility_profile.models import SummaryLog + +from morango.constants import transfer_stages +from morango.constants import transfer_statuses +from morango.models.core import Buffer +from morango.models.core import Store +from morango.models.core import TransferSession +from morango.sync.session import SessionWrapper + +from .helpers import BasePushPullClientTestCase +from .helpers import second_environment + +UUID_RE = re.compile(r"[0-9a-f]{32}") + +# body fields whose values are stable between runs, so their values are pinned along with the keys +PINNED_VALUE_KEYS = ("filter", "push", "records_total", "records_transferred", "transfer_stage") + +# a sync that stalls (for example, waiting on a remote stage that never completes) polls forever, +# so the recorder fails the test once it exceeds either budget +MAX_REQUESTS = 100 +MAX_SECONDS = 120 + +CHUNK_SIZE = 3 + + +def _normalize(value): + if isinstance(value, str): + return UUID_RE.sub("", value) + return value + + +def request(method, path, params=None, encoding=None, keys=(), values=None, retryable=False): + """ + The comparable form of a single request. + + :param method: The HTTP method + :param path: The URL path relative to the server's base URL, with ids replaced by `` + :param params: The query params as a sorted tuple of (name, value) pairs + :param encoding: How the body was sent: "json", "gzip", or None for no body + :param keys: The sorted body keys, or for a list body, the sorted keys of its records + :param values: The values of `PINNED_VALUE_KEYS` in a dict body, as a sorted tuple of pairs, + or for a list body, the number of records + :param retryable: The `is_retryable` flag passed to `SessionWrapper.request` + """ + return (method, path, params, encoding, tuple(keys), values, retryable) + + +class SyncStalledError(AssertionError): + pass + + +class RequestRecorder(object): + """ + Records each request sent through `SessionWrapper.request`, then passes the call through to + the real method so that retries and errors behave as they do without the spy. + """ + + def __init__(self, baseurl): + self.baseurl = baseurl + self.requests = [] + self._original = SessionWrapper.request + self._started = time.monotonic() + + def __call__(self, session, method, url, **kwargs): + self._check_budget() + encoding, body = self._body(kwargs) + self.requests.append( + request( + method.upper(), + self._path(url), + params=self._params(kwargs), + encoding=encoding, + keys=self._keys(body), + values=self._values(body), + retryable=kwargs.get("is_retryable", False), + ) + ) + return self._original(session, method, url, **kwargs) + + def _check_budget(self): + if len(self.requests) >= MAX_REQUESTS: + raise SyncStalledError( + "Sync sent more than {} requests: {}".format(MAX_REQUESTS, self.requests[-5:]) + ) + if time.monotonic() - self._started > MAX_SECONDS: + raise SyncStalledError( + "Sync ran longer than {}s: {}".format(MAX_SECONDS, self.requests[-5:]) + ) + + def _path(self, url): + path = url[len(self.baseurl) :] if url.startswith(self.baseurl) else url + return _normalize(path) + + def _params(self, kwargs): + params = kwargs.get("params") + if not params: + return None + return tuple(sorted((name, _normalize(value)) for name, value in params.items())) + + def _body(self, kwargs): + if kwargs.get("json") is not None: + return "json", kwargs["json"] + data = kwargs.get("data") + if isinstance(data, bytes): + return "gzip", json.loads(gzip.decompress(data).decode("utf-8")) + if data is not None: + return "data", data + return None, None + + def _keys(self, body): + if isinstance(body, dict): + return sorted(body.keys()) + if isinstance(body, list): + # a chunk of serialized buffers, so record the keys shared by its records + return sorted(set().union(*(record.keys() for record in body))) + return () + + def _values(self, body): + if isinstance(body, dict): + return tuple( + sorted((key, _normalize(body[key])) for key in PINNED_VALUE_KEYS if key in body) + ) + if isinstance(body, list): + return len(body) + return None + + @property + def patch(self): + return mock.patch.object(SessionWrapper, "request", autospec=True, side_effect=self) + + +TRANSFER_SESSIONS = "api/morango/v1/transfersessions/" +TRANSFER_SESSION = "api/morango/v1/transfersessions//" +BUFFERS = "api/morango/v1/buffers/" +FILTER = ":user" + +CREATE_TRANSFER_SESSION_KEYS = ("client_fsic", "filter", "id", "push", "sync_session_id") +BUFFER_KEYS = ( + "_self_ref_fk", + "_self_ref_order", + "conflicting_serialized_data", + "deleted", + "hard_deleted", + "last_saved_counter", + "last_saved_instance", + "model_name", + "model_uuid", + "partition", + "profile", + "rmcb_list", + "serialized", + "source_id", + "transfer_session", +) +PULL_TRANSFER_KEYS = ("bytes_received", "bytes_sent", "records_transferred", "transfer_stage") + + +def create_transfer_session(push): + return request( + "POST", + TRANSFER_SESSIONS, + encoding="json", + keys=CREATE_TRANSFER_SESSION_KEYS, + values=(("filter", FILTER), ("push", push)), + retryable=True, + ) + + +def get_transfer_session(): + return request("GET", TRANSFER_SESSION, retryable=True) + + +def update_transfer_session(keys, **values): + return request( + "PATCH", + TRANSFER_SESSION, + encoding="json", + keys=keys, + values=tuple(sorted(values.items())), + retryable=True, + ) + + +def close_transfer_session(): + return request("DELETE", TRANSFER_SESSION, retryable=True) + + +def push_buffers(count): + return request("POST", BUFFERS, encoding="gzip", keys=BUFFER_KEYS, values=count, retryable=True) + + +def pull_buffers(offset): + return request( + "GET", + BUFFERS, + params=(("limit", CHUNK_SIZE), ("offset", offset), ("transfer_session_id", "")), + retryable=True, + ) + + +# 10 records in chunks of 3. Before each stage update, the async flow first reads the remote +# transfer session to check whether the remote is already at that stage. +PUSH_REQUESTS = [ + # initializing + create_transfer_session(push=True), + # serializing + get_transfer_session(), + update_transfer_session( + ("client_fsic", "transfer_stage"), transfer_stage=transfer_stages.SERIALIZING + ), + # queuing + get_transfer_session(), + update_transfer_session( + ("records_total", "transfer_stage"), + records_total=10, + transfer_stage=transfer_stages.QUEUING, + ), + # transferring, where the server advances its transfer session as it receives each chunk + push_buffers(3), + push_buffers(3), + push_buffers(3), + push_buffers(1), + # dequeuing + get_transfer_session(), + update_transfer_session(("transfer_stage",), transfer_stage=transfer_stages.DEQUEUING), + # deserializing + get_transfer_session(), + update_transfer_session(("transfer_stage",), transfer_stage=transfer_stages.DESERIALIZING), + # cleanup + close_transfer_session(), +] + +# 10 records in chunks of 3. Before each stage update, the async flow first reads the remote +# transfer session to check whether the remote is already at that stage. +PULL_REQUESTS = [ + # initializing + create_transfer_session(push=False), + # serializing + get_transfer_session(), + update_transfer_session( + ("client_fsic", "transfer_stage"), transfer_stage=transfer_stages.SERIALIZING + ), + # queuing + get_transfer_session(), + update_transfer_session(("transfer_stage",), transfer_stage=transfer_stages.QUEUING), + # transferring, where the client reports its progress after each chunk + pull_buffers(offset=0), + update_transfer_session( + PULL_TRANSFER_KEYS, records_transferred=3, transfer_stage=transfer_stages.TRANSFERRING + ), + pull_buffers(offset=3), + update_transfer_session( + PULL_TRANSFER_KEYS, records_transferred=6, transfer_stage=transfer_stages.TRANSFERRING + ), + pull_buffers(offset=6), + update_transfer_session( + PULL_TRANSFER_KEYS, records_transferred=9, transfer_stage=transfer_stages.TRANSFERRING + ), + pull_buffers(offset=9), + update_transfer_session( + PULL_TRANSFER_KEYS, records_transferred=10, transfer_stage=transfer_stages.TRANSFERRING + ), + # dequeuing + get_transfer_session(), + update_transfer_session(("transfer_stage",), transfer_stage=transfer_stages.DEQUEUING), + # deserializing + get_transfer_session(), + update_transfer_session(("transfer_stage",), transfer_stage=transfer_stages.DESERIALIZING), + # cleanup + close_transfer_session(), +] + + +class NetworkCharacterizationTestCase(BasePushPullClientTestCase): + def setUp(self): + super(NetworkCharacterizationTestCase, self).setUp() + # the expected requests depend on the chunk size, so pin it here rather than rely on + # the harness + self.conn.chunk_size = CHUNK_SIZE + + def _sync(self, client): + recorder = RequestRecorder(self.server.baseurl) + with recorder.patch: + client.initialize(self.filter) + client.run() + client.finalize() + return recorder.requests + + def _store_count(self): + return Store.objects.filter( + partition__startswith=str(self.filter), profile=self.profile + ).count() + + def _remote_transfer_session(self, transfer_session): + with second_environment(): + return TransferSession.objects.get(id=transfer_session.id) + + def assertTransferSessionFinalState(self, transfer_session, records_total): + transfer_session.refresh_from_db() + self.assertEqual(records_total, transfer_session.records_total) + self.assertEqual(records_total, transfer_session.records_transferred) + self.assertEqual(transfer_stages.CLEANUP, transfer_session.transfer_stage) + self.assertEqual(transfer_statuses.COMPLETED, transfer_session.transfer_stage_status) + self.assertFalse(transfer_session.active) + self.assertEqual(0, Buffer.objects.filter(transfer_session=transfer_session).count()) + + def test_async_push(self): + for _ in range(5): + SummaryLog.objects.create(user=self.local_user) + InteractionLog.objects.create(user=self.local_user) + + client = self.client.get_push_client() + requests = self._sync(client) + self.assertEqual(PUSH_REQUESTS, requests) + + transfer_session = client.context.transfer_session + self.assertTransferSessionFinalState(transfer_session, records_total=10) + + # the client does not report its byte counts to the server during a push + remote_transfer_session = self._remote_transfer_session(transfer_session) + self.assertEqual(0, remote_transfer_session.bytes_sent) + self.assertEqual(0, remote_transfer_session.bytes_received) + + with second_environment(): + # the 10 pushed logs, plus the user + self.assertEqual(11, self._store_count()) + self.assertEqual(5, SummaryLog.objects.filter(user=self.remote_user).count()) + self.assertEqual(5, InteractionLog.objects.filter(user=self.remote_user).count()) + + def test_async_pull(self): + with second_environment(): + for _ in range(5): + SummaryLog.objects.create(user=self.remote_user) + InteractionLog.objects.create(user=self.remote_user) + + client = self.client.get_pull_client() + requests = self._sync(client) + self.assertEqual(PULL_REQUESTS, requests) + + transfer_session = client.context.transfer_session + self.assertTransferSessionFinalState(transfer_session, records_total=10) + + # the client reports its byte counts to the server, flipped to the server's point of view + remote_transfer_session = self._remote_transfer_session(transfer_session) + self.assertEqual(transfer_session.bytes_received, remote_transfer_session.bytes_sent) + self.assertEqual(transfer_session.bytes_sent, remote_transfer_session.bytes_received) + + # the 10 pulled logs, plus the user from the initial pull in `setUp` + self.assertEqual(11, self._store_count()) + self.assertEqual(5, SummaryLog.objects.filter(user=self.local_user).count()) + self.assertEqual(5, InteractionLog.objects.filter(user=self.local_user).count()) From 842b8b69ff3a74dbf5237ac901d06d1d847b43f5 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Thu, 5 Nov 2026 15:18:46 -0800 Subject: [PATCH 3/3] Add characterization tests for legacy network sync Pins what a full push and a full pull send and store when the server lacks ASYNC_OPERATIONS, so the refactors of the LegacyNetwork* operations can show they change neither. The tests run against a live server that emulates a morango version before 0.6.0: the MORANGO_TEST_LEGACY_SERVER variable turns off both ASYNC_OPERATIONS and FSIC_V2_FORMAT on the server, and LiveServer now takes extra environment variables for that. The client keeps its full set of capabilities, as a current Kolibri does when it syncs with an older one. Reducing only the client capabilities, as #364 first proposed, fails today on the server for both push and pull (#416). The expected lists show that the client creates the transfer session and sends no stage updates for serializing, queuing, dequeuing, or deserializing. The legacy pull moves 11 records instead of 10: the client creates the transfer session before it serializes, so it sends an empty client_fsic, and the server queues the user from the initial pull again. Closes #364 Co-Authored-By: Claude Opus 5.5 --- tests/testapp/testapp/server2_settings.py | 7 + tests/testapp/tests/integration/helpers.py | 7 +- .../test_network_characterization.py | 138 +++++++++++++++--- 3 files changed, 126 insertions(+), 26 deletions(-) diff --git a/tests/testapp/testapp/server2_settings.py b/tests/testapp/testapp/server2_settings.py index c7473347..6e43c953 100644 --- a/tests/testapp/testapp/server2_settings.py +++ b/tests/testapp/testapp/server2_settings.py @@ -1,5 +1,12 @@ +import os + from .settings import * # noqa DATABASES = { "default": DATABASES["default2"], # noqa } + +# emulates a server on a morango version before 0.6.0, which lacks both of these capabilities +if os.environ.get("MORANGO_TEST_LEGACY_SERVER"): + MORANGO_DISALLOW_ASYNC_OPERATIONS = True + MORANGO_DISABLE_FSIC_V2_FORMAT = True diff --git a/tests/testapp/tests/integration/helpers.py b/tests/testapp/tests/integration/helpers.py index 62611886..11c72f79 100644 --- a/tests/testapp/tests/integration/helpers.py +++ b/tests/testapp/tests/integration/helpers.py @@ -39,9 +39,10 @@ def second_environment(): class LiveServer: - def __init__(self): + def __init__(self, env=None): self.env = os.environ.copy() self.env["MORANGO_SYSTEM_ID"] = SECOND_SYSTEM_ID + self.env.update(env or {}) self.port = get_free_tcp_port() self.host = "127.0.0.1" self.start() @@ -94,11 +95,13 @@ class BasePushPullClientTestCase(TransactionTestCase): profile = "facilitydata" databases = ["default", SECOND_TEST_DATABASE] + # extra environment variables for the live server process + server_env = {} @classmethod def setUpClass(cls): super(TransactionTestCase, cls).setUpClass() - cls.server = LiveServer() + cls.server = LiveServer(env=cls.server_env) @classmethod def tearDownClass(cls): diff --git a/tests/testapp/tests/integration/test_network_characterization.py b/tests/testapp/tests/integration/test_network_characterization.py index 04dd326f..56fc2fd7 100644 --- a/tests/testapp/tests/integration/test_network_characterization.py +++ b/tests/testapp/tests/integration/test_network_characterization.py @@ -1,6 +1,7 @@ """ -Characterization tests that pin the HTTP requests and the final state of a network push and pull -that use `ASYNC_OPERATIONS`. Refactors of the sync layer must pass these tests without edits. +Characterization tests that pin the HTTP requests and the final state of a network push and pull, +both with `ASYNC_OPERATIONS` and with the legacy flow of a server that lacks it. Refactors of the +sync layer must pass these tests without edits. """ import gzip @@ -14,6 +15,7 @@ from morango.constants import transfer_stages from morango.constants import transfer_statuses +from morango.constants.capabilities import ASYNC_OPERATIONS from morango.models.core import Buffer from morango.models.core import Store from morango.models.core import TransferSession @@ -283,9 +285,55 @@ def pull_buffers(offset): ] -class NetworkCharacterizationTestCase(BasePushPullClientTestCase): +# The legacy server serializes when the client creates the transfer session, and dequeues and +# deserializes when the client closes it, so the client sends no stage updates for those stages. +# The client tells the server how many records it queued, since the server did not queue them. +LEGACY_PUSH_REQUESTS = [ + # initializing + create_transfer_session(push=True), + # queuing + update_transfer_session(("records_total",), records_total=10), + # transferring, where the server advances its transfer session as it receives each chunk + push_buffers(3), + push_buffers(3), + push_buffers(3), + push_buffers(1), + # cleanup + close_transfer_session(), +] + +# The legacy server serializes and queues when the client creates the transfer session, and returns +# the number of records it queued in that response. The client creates the transfer session before +# it serializes, so it sends an empty `client_fsic` and the server queues the user from the initial +# pull in `setUp` again, for 11 records in chunks of 3. +LEGACY_PULL_REQUESTS = [ + # initializing + create_transfer_session(push=False), + # transferring, where the client reports its progress after each chunk + pull_buffers(offset=0), + update_transfer_session( + PULL_TRANSFER_KEYS, records_transferred=3, transfer_stage=transfer_stages.TRANSFERRING + ), + pull_buffers(offset=3), + update_transfer_session( + PULL_TRANSFER_KEYS, records_transferred=6, transfer_stage=transfer_stages.TRANSFERRING + ), + pull_buffers(offset=6), + update_transfer_session( + PULL_TRANSFER_KEYS, records_transferred=9, transfer_stage=transfer_stages.TRANSFERRING + ), + pull_buffers(offset=9), + update_transfer_session( + PULL_TRANSFER_KEYS, records_transferred=11, transfer_stage=transfer_stages.TRANSFERRING + ), + # cleanup + close_transfer_session(), +] + + +class BaseNetworkCharacterizationTestCase(BasePushPullClientTestCase): def setUp(self): - super(NetworkCharacterizationTestCase, self).setUp() + super(BaseNetworkCharacterizationTestCase, self).setUp() # the expected requests depend on the chunk size, so pin it here rather than rely on # the harness self.conn.chunk_size = CHUNK_SIZE @@ -316,15 +364,7 @@ def assertTransferSessionFinalState(self, transfer_session, records_total): self.assertFalse(transfer_session.active) self.assertEqual(0, Buffer.objects.filter(transfer_session=transfer_session).count()) - def test_async_push(self): - for _ in range(5): - SummaryLog.objects.create(user=self.local_user) - InteractionLog.objects.create(user=self.local_user) - - client = self.client.get_push_client() - requests = self._sync(client) - self.assertEqual(PUSH_REQUESTS, requests) - + def _assertPushFinalState(self, client): transfer_session = client.context.transfer_session self.assertTransferSessionFinalState(transfer_session, records_total=10) @@ -339,18 +379,9 @@ def test_async_push(self): self.assertEqual(5, SummaryLog.objects.filter(user=self.remote_user).count()) self.assertEqual(5, InteractionLog.objects.filter(user=self.remote_user).count()) - def test_async_pull(self): - with second_environment(): - for _ in range(5): - SummaryLog.objects.create(user=self.remote_user) - InteractionLog.objects.create(user=self.remote_user) - - client = self.client.get_pull_client() - requests = self._sync(client) - self.assertEqual(PULL_REQUESTS, requests) - + def _assertPullFinalState(self, client, records_total=10): transfer_session = client.context.transfer_session - self.assertTransferSessionFinalState(transfer_session, records_total=10) + self.assertTransferSessionFinalState(transfer_session, records_total=records_total) # the client reports its byte counts to the server, flipped to the server's point of view remote_transfer_session = self._remote_transfer_session(transfer_session) @@ -361,3 +392,62 @@ def test_async_pull(self): self.assertEqual(11, self._store_count()) self.assertEqual(5, SummaryLog.objects.filter(user=self.local_user).count()) self.assertEqual(5, InteractionLog.objects.filter(user=self.local_user).count()) + + def _create_local_logs(self): + for _ in range(5): + SummaryLog.objects.create(user=self.local_user) + InteractionLog.objects.create(user=self.local_user) + + def _create_remote_logs(self): + with second_environment(): + for _ in range(5): + SummaryLog.objects.create(user=self.remote_user) + InteractionLog.objects.create(user=self.remote_user) + + +class NetworkCharacterizationTestCase(BaseNetworkCharacterizationTestCase): + def test_async_push(self): + self._create_local_logs() + + client = self.client.get_push_client() + self.assertIn(ASYNC_OPERATIONS, client.context.capabilities) + requests = self._sync(client) + self.assertEqual(PUSH_REQUESTS, requests) + self._assertPushFinalState(client) + + def test_async_pull(self): + self._create_remote_logs() + + client = self.client.get_pull_client() + self.assertIn(ASYNC_OPERATIONS, client.context.capabilities) + requests = self._sync(client) + self.assertEqual(PULL_REQUESTS, requests) + self._assertPullFinalState(client) + + +class LegacyNetworkCharacterizationTestCase(BaseNetworkCharacterizationTestCase): + """ + Syncs with a server that emulates a morango version before 0.6.0, so the server lacks + `ASYNC_OPERATIONS` and `FSIC_V2_FORMAT`. The client keeps its full set of capabilities. + """ + + server_env = {"MORANGO_TEST_LEGACY_SERVER": "1"} + + def test_legacy_push(self): + self._create_local_logs() + + client = self.client.get_push_client() + self.assertNotIn(ASYNC_OPERATIONS, client.context.capabilities) + requests = self._sync(client) + self.assertEqual(LEGACY_PUSH_REQUESTS, requests) + self._assertPushFinalState(client) + + def test_legacy_pull(self): + self._create_remote_logs() + + client = self.client.get_pull_client() + self.assertNotIn(ASYNC_OPERATIONS, client.context.capabilities) + requests = self._sync(client) + self.assertEqual(LEGACY_PULL_REQUESTS, requests) + # the server sent the user from the initial pull again, which the client already had + self._assertPullFinalState(client, records_total=11)