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
new file mode 100644
index 00000000..11c72f79
--- /dev/null
+++ b/tests/testapp/tests/integration/helpers.py
@@ -0,0 +1,195 @@
+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, 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()
+
+    @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]
+    # extra environment variables for the live server process
+    server_env = {}
+
+    @classmethod
+    def setUpClass(cls):
+        super(TransactionTestCase, cls).setUpClass()
+        cls.server = LiveServer(env=cls.server_env)
+
+    @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_network_characterization.py b/tests/testapp/tests/integration/test_network_characterization.py
new file mode 100644
index 00000000..56fc2fd7
--- /dev/null
+++ b/tests/testapp/tests/integration/test_network_characterization.py
@@ -0,0 +1,453 @@
+"""
+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
+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.constants.capabilities import ASYNC_OPERATIONS
+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("<id>", 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 `<id>`
+    :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/<id>/"
+BUFFERS = "api/morango/v1/buffers/"
+FILTER = "<id>: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", "<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(),
+]
+
+
+# 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(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
+
+    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 _assertPushFinalState(self, client):
+        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 _assertPullFinalState(self, client, records_total=10):
+        transfer_session = client.context.transfer_session
+        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)
+        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())
+
+    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)
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: