From 8ecbe975cd648f38356cf35ad8d4b4c926e43269 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <matthias@neon.tech>
Date: Thu, 12 Sep 2024 23:34:20 +0100
Subject: [PATCH] Intermediate state for process-local Endpoints when accessing
 shared state

---
 test_runner/fixtures/shared_fixtures.py | 66 +++++++++++++++++++++++++
 1 file changed, 66 insertions(+)
 create mode 100644 test_runner/fixtures/shared_fixtures.py

diff --git a/test_runner/fixtures/shared_fixtures.py b/test_runner/fixtures/shared_fixtures.py
new file mode 100644
index 0000000000..752c5597ae
--- /dev/null
+++ b/test_runner/fixtures/shared_fixtures.py
@@ -0,0 +1,66 @@
+from typing import Optional, List, Dict, Any
+
+from fixtures.common_types import TimelineId, Lsn
+from fixtures.neon_fixtures import PgProtocol, Safekeeper
+
+
+class TestEndpoint(PgProtocol):
+    def __init__(self, **kwargs: Any):
+        super().__init__(**kwargs)
+
+    def start(self):
+        pass
+
+class TestTimeline:
+    primary: Optional[TestEndpoint]
+    secondaries: List[TestEndpoint]
+
+    def __init__(self):
+        self.primary = None
+
+    def start_primary(self) -> Optional[TestEndpoint]:
+        if not self.primary:
+            return None
+        self.primary.start()
+
+        return self.primary
+
+class TestPageserver:
+    pass
+
+class TestSafekeeper:
+    pass
+
+class TestTenant:
+    """
+    An object representing a single test case on a shared pageserver.
+    All operations here are safe practically safe.
+    """
+
+    def __init__(
+        self,
+        pageserver: TestPageserver,
+        safekeeper: Safekeeper,
+        main_timeline_name: Optional[str] = "main",
+    ):
+        self.pageserver = pageserver
+        self.safekeeper = safekeeper
+        self.timeline_map: Dict[str, TestTimeline] = dict()
+
+        self.timeline_map[main_timeline_name] = self.primary_timeline
+
+    def create_timeline(
+        self,
+        name: str,
+        parent_name: Optional[str],
+        parent_id: Optional[TimelineId],
+        branch_at: Optional[Lsn],
+    ) -> TestTimeline:
+        if (tlid := self.timeline_map.get(name)) is not None:
+            return tlid
+
+        return self._new_timeline()
+
+    def _new_timeline(self) -> TestTimeline:
+        pass
+