diff --git a/octobot/community/authentication.py b/octobot/community/authentication.py
index d6675baa1f..add3efcce5 100644
--- a/octobot/community/authentication.py
+++ b/octobot/community/authentication.py
@@ -49,6 +49,8 @@
import octobot_trading.enums as trading_enums
import octobot_sync.client as sync_client
import octobot_sync.chain as sync_chain
+import octobot_sync.mirror.writer as sync_session_writer
+import starfish_spaces
def expired_session_retrier(func):
@@ -133,6 +135,8 @@ def __init__(self, config=None, backend_url=None, backend_key=None, use_as_singl
self._sync_client = None
self.sync_user_id: str = ""
self._sync_client_lock = threading.Lock()
+ self._dk_sessions: dict[str, starfish_spaces.Session] = {}
+ self._dk_session_lock = asyncio.Lock()
self._wallet_backend: wallet_backend.WalletBackend = wallet_backend.WalletBackend(
self._get_wallet_sync_storage(), self.logger
)
@@ -607,6 +611,10 @@ async def stop(self):
if self._sync_client:
await self._sync_client.close()
self._sync_client = None
+ for session in self._dk_sessions.values():
+ await session.content_client.close()
+ await session.account_client.close()
+ self._dk_sessions.clear()
self.logger.debug("Stopped")
def _update_supports(self, resp_status, json_data):
@@ -712,6 +720,30 @@ def get_wallet(self, address: str) -> sync_chain.Wallet:
def get_wallet_by_user_id(self, user_id: str) -> sync_chain.Wallet:
return self._wallet_backend.get_wallet_by_user_id(user_id)
+ async def get_session_for_address(self, address: str) -> starfish_spaces.Session:
+ """Build (and cache) a dk-namespace starfish_spaces Session for the given wallet.
+
+ Used for trading-signal publish/pull — see octobot_sync.artifacts. Distinct from
+ get_sync_client_for_address's bare StarfishClient: a space needs a Session.
+ """
+ cached = self._dk_sessions.get(address)
+ if cached is not None:
+ return cached
+ async with self._dk_session_lock:
+ cached = self._dk_sessions.get(address)
+ if cached is not None:
+ return cached
+ sync_url = identifiers_provider.IdentifiersProvider.SYNC_SERVER_URL
+ if not sync_url:
+ raise wallet_backend.WalletError("No sync server URL configured")
+ wallet = self.get_wallet(address)
+ derived = sync_session_writer.derived_identity_for_mirror(wallet.private_key)
+ session = await sync_session_writer.build_mirror_session(
+ derived, sync_url, name="octobot-signals"
+ )
+ self._dk_sessions[address] = session
+ return session
+
def init_sync_client_for_wallet(self, address: str) -> None:
"""Initialize the sync client for the given wallet address without passphrase."""
if self._sync_client is not None:
diff --git a/octobot/constants.py b/octobot/constants.py
index 916bfa2c08..d85952686a 100644
--- a/octobot/constants.py
+++ b/octobot/constants.py
@@ -110,8 +110,9 @@
OCTOBOT_MARKET_MAKING_URL = os.getenv("OCTOBOT_MARKET_MAKING_URL", "https://market-making.octobot.cloud")
# sync server
-SYNC_SERVER_URL = os.getenv("SYNC_SERVER_URL", "https://prod-sync.drakkar.software/sync")
-STAGING_SYNC_SERVER_URL = os.getenv("SYNC_SERVER_URL", "https://beta-sync.drakkar.software/sync")
+# Bare origin: octobot_sync.client/mirror.writer append SYNC_MOUNT_PATH ("sync") themselves.
+SYNC_SERVER_URL = os.getenv("SYNC_SERVER_URL", "https://prod-sync.drakkar.software")
+STAGING_SYNC_SERVER_URL = os.getenv("SYNC_SERVER_URL", "https://beta-sync.drakkar.software")
SYNC_NAMESPACE=os.getenv("SYNC_NAMESPACE", "dk")
SYNC_CHAIN_ID = os.getenv("SYNC_CHAIN_ID", "evm:8453")
diff --git a/packages/flow/octobot_flow/repositories/community/community_repository.py b/packages/flow/octobot_flow/repositories/community/community_repository.py
index 52f764ee41..90cde070bf 100644
--- a/packages/flow/octobot_flow/repositories/community/community_repository.py
+++ b/packages/flow/octobot_flow/repositories/community/community_repository.py
@@ -2,6 +2,8 @@
import asyncio
import typing
+import starfish_spaces
+
import octobot.community
import octobot_commons.logging
@@ -67,3 +69,10 @@ def _get_sync_client(self) -> octobot_sync.client.StarfishClient:
"Wallet not initialized: no wallet address provided"
)
return self.authenticator.get_sync_client_for_address(self.wallet_address)
+
+ async def _get_signal_session(self) -> starfish_spaces.Session:
+ if self.wallet_address is None:
+ raise octobot_flow.errors.WalletNotInitializedError(
+ "Wallet not initialized: no wallet address provided"
+ )
+ return await self.authenticator.get_session_for_address(self.wallet_address)
diff --git a/packages/flow/octobot_flow/repositories/community/trading_signals_repository.py b/packages/flow/octobot_flow/repositories/community/trading_signals_repository.py
index 753d312e29..529de37593 100644
--- a/packages/flow/octobot_flow/repositories/community/trading_signals_repository.py
+++ b/packages/flow/octobot_flow/repositories/community/trading_signals_repository.py
@@ -1,33 +1,26 @@
import dataclasses
import typing
+import starfish_spaces
+
import octobot_commons.dataclasses
import octobot_commons.json_util
import octobot_commons.constants
import octobot_commons.logging as logging
-import octobot_sync.client
+import octobot_sync.artifacts
import octobot_flow.entities
+import octobot_flow.errors
import octobot_flow.repositories.community.trading_signals_channel as trading_signals_channel
import octobot_flow.repositories.community.community_repository as community_repository
-import octobot_flow.errors
import octobot_flow.constants
+# artifact-events "version" path segment: a fixed wire-schema version, not a strategy revision.
VERSION = "1.0.0"
-def _trading_signal_fingerprint(signal: octobot_flow.entities.TradingSignal) -> tuple[str, float]:
- """Stable identity for dedupe across JSON round-trips (float normalizes Decimal)."""
- return (signal.strategy_id, float(signal.account.updated_at))
-
-
-def _signals_sorted_chronologically(
- signals: list[octobot_flow.entities.TradingSignal],
-) -> list[octobot_flow.entities.TradingSignal]:
- """Sort by ``account.updated_at``, then original index when timestamps tie."""
- indexed = list(enumerate(signals))
- indexed.sort(key=lambda index_signal: (float(index_signal[1].account.updated_at), index_signal[0]))
- return [signal for _, signal in indexed]
+def _signal_space_id(strategy_id: str) -> str:
+ return octobot_sync.artifacts.artifact_space_id(f"octobot-signals-{strategy_id}")
@dataclasses.dataclass
@@ -41,34 +34,6 @@ def __post_init__(self):
for signal in self.signals
]
- def merge_with_remote(self, remote: "TradingSignalPayload") -> "TradingSignalPayload":
- """Combine server history with client-only snapshots for optimistic-concurrency merge."""
- remote_ordered = _signals_sorted_chronologically(remote.signals)
- remote_fingerprints = {_trading_signal_fingerprint(signal) for signal in remote_ordered}
- local_ordered = _signals_sorted_chronologically(self.signals)
- local_only = [
- signal
- for signal in local_ordered
- if _trading_signal_fingerprint(signal) not in remote_fingerprints
- ]
- return TradingSignalPayload(signals=remote_ordered + local_only)
-
-
-def _sync_signals_path(sync_kind: str, strategy_id: str) -> str:
- return f"/v1/{sync_kind}/products/{strategy_id}/{VERSION}/signals"
-
-
-def _merge_trading_signal_documents(
- local_payload: dict[str, typing.Any],
- remote_payload: dict[str, typing.Any],
-) -> dict[str, typing.Any]:
- """Merge pending client document with server state after a sync push conflict (409)."""
- local_model = TradingSignalPayload.from_dict(local_payload if isinstance(local_payload, dict) else {})
- remote_model = TradingSignalPayload.from_dict(remote_payload if isinstance(remote_payload, dict) else {})
- return octobot_commons.json_util.sanitize(
- local_model.merge_with_remote(remote_model).to_dict()
- )
-
def _trim_historical_snapshots_if_needed(
trading_signal: octobot_flow.entities.TradingSignal,
@@ -91,10 +56,15 @@ async def fetch_trading_signals(
history_size: int,
) -> list[octobot_flow.entities.TradingSignal]:
trading_signals: list[octobot_flow.entities.TradingSignal] = []
+ session: typing.Optional[starfish_spaces.Session] = None
+ failed_strategy_ids: list[str] = []
+ first_failure: typing.Optional[Exception] = None
for strategy_identifier in strategy_ids:
try:
+ if session is None:
+ session = await self._get_signal_session()
pulled_signals = await self._pull_trading_signals(
- self._get_sync_client(), strategy_identifier, history_size
+ session, strategy_identifier, history_size
)
if not pulled_signals.signals:
continue
@@ -110,36 +80,46 @@ async def fetch_trading_signals(
True,
f"Failed to fetch trading signals for strategy {strategy_identifier!r}: {strategy_error}",
)
+ failed_strategy_ids.append(strategy_identifier)
+ if first_failure is None:
+ first_failure = strategy_error
+ if failed_strategy_ids:
+ raise octobot_flow.errors.CommunityTradingSignalError(
+ f"Failed to fetch trading signals for {', '.join(failed_strategy_ids)}: {first_failure}"
+ ) from first_failure
return trading_signals
async def _upload_trading_signal(
self,
trading_signal: octobot_flow.entities.TradingSignal,
):
- client = self._get_sync_client()
payload = octobot_commons.json_util.sanitize(trading_signal.to_dict())
try:
- await octobot_sync.client.append_payload(
- client,
- push_path=_sync_signals_path("push", trading_signal.strategy_id),
- payload=payload,
- timestamp=int(trading_signal.account.updated_at * octobot_commons.constants.MSECONDS_TO_SECONDS),
+ session = await self._get_signal_session()
+ await octobot_sync.artifacts.publish_artifact_event(
+ session,
+ _signal_space_id(trading_signal.strategy_id),
+ VERSION,
+ payload,
+ ts=int(trading_signal.account.updated_at * octobot_commons.constants.MSECONDS_TO_SECONDS),
)
except Exception as upload_error:
self._logger().exception(upload_error, True, f"Failed to upload trading signal: {upload_error}")
async def _pull_trading_signals(
- self, client: octobot_sync.client.StarfishClient, strategy_id: str, last: typing.Optional[int]
+ self,
+ session: starfish_spaces.Session,
+ strategy_id: str,
+ last: typing.Optional[int],
+ *,
+ owner_ed_pub: typing.Optional[str] = None,
) -> TradingSignalPayload:
- signals = await client.pull(
- _sync_signals_path("pull", strategy_id),
- last=last
+ signals = await octobot_sync.artifacts.pull_artifact_events(
+ session, _signal_space_id(strategy_id), VERSION, last, owner_ed_pub=owner_ed_pub
)
- if not isinstance(signals, list):
- raise octobot_flow.errors.CommunityTradingSignalError(f"Unexpected response type: {type(signals)}")
return TradingSignalPayload(
signals=[
- octobot_flow.entities.TradingSignal.from_dict(signal["data"]) for signal in signals
+ octobot_flow.entities.TradingSignal.from_dict(signal) for signal in signals
]
)
diff --git a/packages/flow/tests/functionnal_tests/community/test_trading_signals_repository.py b/packages/flow/tests/functionnal_tests/community/test_trading_signals_repository.py
new file mode 100644
index 0000000000..e2deb007c5
--- /dev/null
+++ b/packages/flow/tests/functionnal_tests/community/test_trading_signals_repository.py
@@ -0,0 +1,304 @@
+import re
+
+import httpx
+import mock
+import pytest
+import pytest_asyncio
+from fastapi import FastAPI
+
+from starfish_sdk import StarfishClient
+from starfish_server.config.schema import (
+ SyncConfig,
+ CollectionConfig,
+ NamespaceConfig,
+ AppendOnlyConfig,
+)
+from starfish_sharing import make_registry_role_enricher
+from starfish_spaces.client import ClientOpts, DeviceKeys
+from starfish_spaces.session import BuildSessionOpts, build_session
+
+import octobot_copy.constants as copy_constants
+import octobot_protocol.models as protocol_models
+import octobot_sync.app as sync_app
+import octobot_sync.artifacts as artifacts
+import octobot_sync.constants as sync_constants
+
+import octobot_flow.entities
+import octobot_flow.errors
+import octobot_flow.repositories.community.trading_signals_repository as trading_signals_repository
+
+
+class _MemoryObjectStore:
+ """Minimal in-memory AbstractObjectStore (every method takes *, context=None). Mirrors
+ packages/sync/tests/test_integration_artifacts.py's fixture of the same shape."""
+
+ def __init__(self) -> None:
+ self._store: dict[str, str] = {}
+
+ async def get_string(self, key, *, context=None):
+ return self._store.get(key)
+
+ async def put(self, key, body, *, content_type=None, cache_control=None, context=None):
+ self._store[key] = body
+
+ async def list_keys(self, prefix, *, start_after=None, limit=None, context=None):
+ keys = sorted(k for k in self._store if k.startswith(prefix))
+ if start_after:
+ keys = [k for k in keys if k > start_after]
+ return keys[:limit] if limit else keys
+
+ async def delete(self, key, *, context=None):
+ self._store.pop(key, None)
+
+ async def delete_many(self, keys, *, context=None):
+ for k in keys:
+ self._store.pop(k, None)
+
+
+_DK_NAMESPACE = "dk"
+
+# Same dk_spaces collection set as packages/sync/tests/test_integration_artifacts.py — kept
+# separate (not cross-package imported) since this package's tests aren't on the same sys.path
+# root as packages/sync's.
+_DK_CONFIG = SyncConfig(
+ version=1,
+ collections=[],
+ namespaces={
+ _DK_NAMESPACE: NamespaceConfig(
+ collections=[
+ CollectionConfig(
+ name="spaceregistry",
+ storagePath="spaces/{spaceId}/_access",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="none",
+ maxBodyBytes=131_072,
+ ),
+ CollectionConfig(
+ name="spacekeyring",
+ storagePath="spaces/{spaceId}/_keyring",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="none",
+ maxBodyBytes=65_536,
+ ),
+ CollectionConfig(
+ name="artifact-events",
+ storagePath="spaces/{spaceId}/artifact/versions/{version}/events",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="delegated",
+ appendOnly=AppendOnlyConfig(type="by_timestamp", requireAuthorSignature=True),
+ maxBodyBytes=sync_constants.MAX_BODY_SIZE_SIGNAL,
+ ),
+ ]
+ ),
+ },
+)
+
+
+def _make_dk_role_enricher(store: _MemoryObjectStore):
+ return make_registry_role_enricher(
+ store,
+ id_param="spaceId",
+ registry_path="spaces/{id}/_access",
+ owner_role="space:owner",
+ member_role="space:member",
+ allow_tofu=True,
+ id_pattern=re.compile(r"^[a-zA-Z0-9_-]+$"),
+ )
+
+
+# Well-known Anvil/Hardhat account #1 — public, safe to embed in tests.
+_PRIV = "0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d"
+
+
+@pytest.fixture
+def dk_app():
+ store = _MemoryObjectStore()
+ inner = sync_app.create_app(
+ store, sync_config=_DK_CONFIG, role_enricher=_make_dk_role_enricher(store)
+ )
+ outer = FastAPI()
+ outer.mount("/sync", inner)
+ return outer
+
+
+async def _build_session(dk_app):
+ """A real starfish_spaces Session wired to the in-process dk_app via ASGITransport — same
+ transport-monkeypatch pattern as packages/sync/tests/test_integration_artifacts.py."""
+ transport = httpx.ASGITransport(app=dk_app)
+
+ from octobot_sync.auth.provider import derive_root_identity
+ import starfish_spaces.client as spaces_client_module
+ import starfish_spaces.session as spaces_session_module
+
+ root = derive_root_identity(_PRIV)
+ keys: DeviceKeys = {
+ "edPriv": root.keys.ed_priv,
+ "edPub": root.keys.ed_pub,
+ "kemPriv": root.keys.kem_priv,
+ "kemPub": root.keys.kem_pub,
+ }
+ client_opts: ClientOpts = {"baseUrl": "http://test/sync", "namespace": _DK_NAMESPACE}
+
+ def _make_space_client(cap, ed_priv_hex, opts):
+ return StarfishClient(
+ opts["baseUrl"],
+ cap_provider=spaces_client_module.cap_provider_for(cap, ed_priv_hex),
+ namespace=opts.get("namespace"),
+ timeout=float(opts.get("timeout", 30.0)),
+ client=httpx.AsyncClient(transport=transport),
+ )
+
+ original = spaces_session_module.make_space_client
+ spaces_session_module.make_space_client = _make_space_client
+ try:
+ return await build_session(
+ BuildSessionOpts(user_id=root.user_id, keys=keys, client_opts=client_opts)
+ )
+ finally:
+ spaces_session_module.make_space_client = original
+
+
+@pytest_asyncio.fixture
+async def signal_session(dk_app):
+ session = await _build_session(dk_app)
+ yield session
+ await session.content_client.close()
+ await session.account_client.close()
+
+
+@pytest.fixture
+def repository(signal_session):
+ """A TradingSignalsRepository whose _get_signal_session returns a real, in-process
+ dk_spaces Session — so _upload_trading_signal/fetch_trading_signals exercise the real
+ octobot_sync.artifacts calls, not a mock, matching the pattern already used for
+ _upload_trading_signal in test_trading_signals_channel.py."""
+ repo = trading_signals_repository.TradingSignalsRepository(mock.MagicMock())
+ repo._get_signal_session = mock.AsyncMock(return_value=signal_session)
+ return repo
+
+
+def _trading_signal(strategy_id: str, updated_at: float) -> octobot_flow.entities.TradingSignal:
+ return octobot_flow.entities.TradingSignal(
+ strategy_id=strategy_id,
+ account=protocol_models.CopiedAccount(
+ version=copy_constants.COPIED_ACCOUNT_VERSION,
+ updated_at=updated_at,
+ copied_assets=[],
+ ),
+ )
+
+
+class TestSignalSpaceId:
+ def test_is_deterministic_for_the_same_strategy_id(self):
+ assert trading_signals_repository._signal_space_id(
+ "strat-a"
+ ) == trading_signals_repository._signal_space_id("strat-a")
+
+ def test_differs_across_strategy_ids(self):
+ assert trading_signals_repository._signal_space_id(
+ "strat-a"
+ ) != trading_signals_repository._signal_space_id("strat-b")
+
+
+class TestUploadAndFetchTradingSignal:
+ @pytest.mark.asyncio
+ async def test_upload_then_fetch_round_trips_a_trading_signal(self, repository):
+ # The real gap this closes: _upload_trading_signal and _pull_trading_signals each
+ # compute their own space_id via _signal_space_id(strategy_id) — nothing outside
+ # this repository proves those two computations agree for the same strategy_id.
+ signal = _trading_signal("strat-roundtrip", 1000.0)
+
+ await repository._upload_trading_signal(signal)
+ fetched = await repository.fetch_trading_signals(["strat-roundtrip"], history_size=10)
+
+ assert len(fetched) == 1
+ assert fetched[0].strategy_id == "strat-roundtrip"
+ assert fetched[0].account.updated_at == 1000.0
+
+ @pytest.mark.asyncio
+ async def test_fetch_of_an_unpublished_strategy_returns_nothing(self, repository):
+ fetched = await repository.fetch_trading_signals(["never-published"], history_size=10)
+ assert fetched == []
+
+ @pytest.mark.asyncio
+ async def test_upload_trading_signal_swallows_publish_errors(self, repository):
+ with mock.patch.object(
+ artifacts, "publish_artifact_event", mock.AsyncMock(side_effect=RuntimeError("boom"))
+ ):
+ # Must not raise — insert_trading_signal's caller (e.g. an automation job) must
+ # not crash because the sync server was unreachable.
+ await repository._upload_trading_signal(_trading_signal("strat-x", 1.0))
+
+ @pytest.mark.asyncio
+ async def test_fetch_trading_signals_raises_for_a_failing_strategy_after_trying_the_rest(
+ self, repository
+ ):
+ ok_signal = _trading_signal("strat-ok", 1000.0)
+ await repository._upload_trading_signal(ok_signal)
+
+ broken_space_id = trading_signals_repository._signal_space_id("strat-broken")
+ real_pull = artifacts.pull_artifact_events
+ attempted_space_ids = []
+
+ async def _pull_that_fails_for_the_broken_space(session, space_id, version, last, **kwargs):
+ attempted_space_ids.append(space_id)
+ if space_id == broken_space_id:
+ raise RuntimeError("boom")
+ return await real_pull(session, space_id, version, last, **kwargs)
+
+ with mock.patch.object(
+ artifacts, "pull_artifact_events", _pull_that_fails_for_the_broken_space
+ ):
+ with pytest.raises(octobot_flow.errors.CommunityTradingSignalError, match="strat-broken"):
+ await repository.fetch_trading_signals(
+ ["strat-broken", "strat-ok"], history_size=10
+ )
+
+ # Both strategies were attempted — the broken one didn't abort the loop early.
+ ok_space_id = trading_signals_repository._signal_space_id("strat-ok")
+ assert set(attempted_space_ids) == {broken_space_id, ok_space_id}
+
+ @pytest.mark.asyncio
+ async def test_fetch_trading_signals_reuses_a_single_session_across_strategies(self, repository):
+ await repository._upload_trading_signal(_trading_signal("strat-a", 1.0))
+ await repository._upload_trading_signal(_trading_signal("strat-b", 2.0))
+ repository._get_signal_session.reset_mock()
+
+ fetched = await repository.fetch_trading_signals(["strat-a", "strat-b"], history_size=10)
+
+ assert {signal.strategy_id for signal in fetched} == {"strat-a", "strat-b"}
+ repository._get_signal_session.assert_awaited_once()
+
+ @pytest.mark.asyncio
+ async def test_fetch_trading_signals_returns_the_most_recently_updated_signal(self, repository):
+ strategy_id = "strat-newest-wins"
+ await repository._upload_trading_signal(_trading_signal(strategy_id, 100.0))
+ await repository._upload_trading_signal(_trading_signal(strategy_id, 200.0))
+
+ fetched = await repository.fetch_trading_signals([strategy_id], history_size=10)
+
+ assert len(fetched) == 1
+ assert fetched[0].account.updated_at == 200.0
+
+ @pytest.mark.asyncio
+ async def test_fetch_trading_signals_trims_historical_snapshots_to_history_size(self, repository):
+ strategy_id = "strat-trim"
+ snapshots = [
+ protocol_models.CopiedAccount(
+ version=copy_constants.COPIED_ACCOUNT_VERSION,
+ updated_at=float(index),
+ copied_assets=[],
+ )
+ for index in range(5)
+ ]
+ signal = _trading_signal(strategy_id, 1000.0)
+ signal.account.historical_snapshots = snapshots
+
+ await repository._upload_trading_signal(signal)
+ fetched = await repository.fetch_trading_signals([strategy_id], history_size=2)
+
+ assert len(fetched) == 1
+ assert len(fetched[0].account.historical_snapshots) == 2
diff --git a/packages/flow/tests/repositories/community/test_community_repository.py b/packages/flow/tests/repositories/community/test_community_repository.py
index 36ccee7914..9b28507710 100644
--- a/packages/flow/tests/repositories/community/test_community_repository.py
+++ b/packages/flow/tests/repositories/community/test_community_repository.py
@@ -1,5 +1,7 @@
import mock
+import pytest
+import octobot_flow.errors
import octobot_flow.repositories.community.community_repository as community_repository_module
@@ -35,3 +37,22 @@ def test_user_id_to_evm_returns_none_when_resolution_fails(self):
result = community_repository_module.CommunityRepository.user_id_to_evm("user_1")
assert result is None
mock_get_logger.return_value.warning.assert_called_once()
+
+
+class TestGetSignalSession:
+ @pytest.mark.asyncio
+ async def test_raises_when_wallet_address_is_none(self):
+ repository = community_repository_module.CommunityRepository(mock.MagicMock(), wallet_address=None)
+ with pytest.raises(octobot_flow.errors.WalletNotInitializedError):
+ await repository._get_signal_session()
+
+ @pytest.mark.asyncio
+ async def test_forwards_wallet_address_to_the_authenticator(self):
+ sentinel_session = mock.Mock()
+ authenticator = mock.Mock(get_session_for_address=mock.AsyncMock(return_value=sentinel_session))
+ repository = community_repository_module.CommunityRepository(authenticator, wallet_address="0xabc")
+
+ result = await repository._get_signal_session()
+
+ assert result is sentinel_session
+ authenticator.get_session_for_address.assert_awaited_once_with("0xabc")
diff --git a/packages/flow/tests/repositories/community/test_trading_signals_channel.py b/packages/flow/tests/repositories/community/test_trading_signals_channel.py
index 3f35155c6b..85cfd500d4 100644
--- a/packages/flow/tests/repositories/community/test_trading_signals_channel.py
+++ b/packages/flow/tests/repositories/community/test_trading_signals_channel.py
@@ -1,6 +1,4 @@
import asyncio
-import decimal
-import json
import mock
import pytest
import pytest_asyncio
@@ -8,7 +6,6 @@
import async_channel.channels as async_channel_channels
-import octobot_commons.json_util as json_util_module
import octobot_copy.constants as copy_constants
import octobot_protocol.models as protocol_models
@@ -86,132 +83,3 @@ async def test_shutdown_internal_trading_signal_channel_allows_recreate(internal
new_channel = await trading_signals_channel.get_or_create_internal_trading_signal_channel()
assert new_channel is not None
-
-def _trading_signal(strategy_id: str, updated_at: float) -> octobot_flow.entities.TradingSignal:
- return octobot_flow.entities.TradingSignal(
- strategy_id=strategy_id,
- account=protocol_models.CopiedAccount(
- version=copy_constants.COPIED_ACCOUNT_VERSION,
- updated_at=updated_at,
- copied_assets=[],
- ),
- )
-
-
-def _payload_dict(*signals: octobot_flow.entities.TradingSignal) -> dict:
- return trading_signals_repository.TradingSignalPayload(signals=list(signals)).to_dict()
-
-
-def _merge_result_updated_at_sequence(merged_dict: dict) -> list[float]:
- parsed = trading_signals_repository.TradingSignalPayload.from_dict(merged_dict)
- return [float(signal.account.updated_at) for signal in parsed.signals]
-
-
-def _assert_no_decimal_leaves(value) -> None:
- """``_merge_trading_signal_documents`` wraps the merged dict in ``json_util.sanitize`` (no nested Decimal)."""
- if isinstance(value, dict):
- for nested in value.values():
- _assert_no_decimal_leaves(nested)
- elif isinstance(value, (list, tuple)):
- for item in value:
- _assert_no_decimal_leaves(item)
- else:
- assert not isinstance(value, decimal.Decimal), f"expected sanitized merge, got Decimal: {value!r}"
-
-
-class TestMergeTradingSignalDocuments:
- """Covers ``_merge_trading_signal_documents``; results are ``json_util.sanitize(merged.to_dict())`` for Starfish push."""
-
- def test_both_empty_documents_yield_empty_signals(self):
- merged = trading_signals_repository._merge_trading_signal_documents({}, {})
- assert merged == json_util_module.sanitize({"signals": []})
- _assert_no_decimal_leaves(merged)
- json.dumps(merged)
-
- def test_empty_remote_yields_local_signals_sorted_chronologically(self):
- first_later = _trading_signal("s", 200.0)
- second_earlier = _trading_signal("s", 100.0)
- merged = trading_signals_repository._merge_trading_signal_documents(
- _payload_dict(first_later, second_earlier),
- {},
- )
- assert _merge_result_updated_at_sequence(merged) == [100.0, 200.0]
- _assert_no_decimal_leaves(merged)
- json.dumps(merged)
-
- def test_empty_local_yields_remote_signals_sorted_chronologically(self):
- remote_third = _trading_signal("s", 300.0)
- remote_first = _trading_signal("s", 100.0)
- remote_second = _trading_signal("s", 200.0)
- merged = trading_signals_repository._merge_trading_signal_documents(
- {},
- _payload_dict(remote_third, remote_first, remote_second),
- )
- assert _merge_result_updated_at_sequence(merged) == [100.0, 200.0, 300.0]
- _assert_no_decimal_leaves(merged)
- json.dumps(merged)
-
- def test_deduplicates_remote_then_appends_new_local_snapshots(self):
- remote_a = _trading_signal("strat", 1.0)
- remote_b = _trading_signal("strat", 2.0)
- local_a = _trading_signal("strat", 1.0)
- local_b = _trading_signal("strat", 2.0)
- local_c = _trading_signal("strat", 3.0)
- merged = trading_signals_repository._merge_trading_signal_documents(
- _payload_dict(local_a, local_b, local_c),
- _payload_dict(remote_a, remote_b),
- )
- assert _merge_result_updated_at_sequence(merged) == [1.0, 2.0, 3.0]
- _assert_no_decimal_leaves(merged)
- json.dumps(merged)
-
- def test_stable_order_when_two_signals_share_timestamp(self):
- first_at_tie = _trading_signal("strat", 50.0)
- second_at_tie = _trading_signal("strat", 50.0)
- merged = trading_signals_repository._merge_trading_signal_documents(
- {},
- _payload_dict(second_at_tie, first_at_tie),
- )
- sequence = _merge_result_updated_at_sequence(merged)
- assert sequence == [50.0, 50.0]
- _assert_no_decimal_leaves(merged)
- json.dumps(merged)
-
- def test_local_only_tail_sorted_when_new_signals_not_in_remote(self):
- remote_lo = _trading_signal("strat", 10.0)
- remote_hi = _trading_signal("strat", 20.0)
- local_newer_first_in_list = _trading_signal("strat", 50.0)
- local_newer_second_in_list = _trading_signal("strat", 40.0)
- merged = trading_signals_repository._merge_trading_signal_documents(
- _payload_dict(remote_lo, remote_hi, local_newer_first_in_list, local_newer_second_in_list),
- _payload_dict(remote_lo, remote_hi),
- )
- assert _merge_result_updated_at_sequence(merged) == [10.0, 20.0, 40.0, 50.0]
- _assert_no_decimal_leaves(merged)
- json.dumps(merged)
-
- def test_merge_with_portfolio_decimals_is_sanitized(self):
- """Mirrors sync JSON: copied_asset totals may be Decimals in-domain but must become floats after merge."""
- account_with_decimals = protocol_models.CopiedAccount(
- version=copy_constants.COPIED_ACCOUNT_VERSION,
- updated_at=1.0,
- copied_assets=[
- protocol_models.CopiedAsset(
- name="USDC",
- total=float(decimal.Decimal("1000.5")),
- available=float(decimal.Decimal("1000.5")),
- ratio=1.0,
- ),
- ],
- )
- signal = octobot_flow.entities.TradingSignal(strategy_id="s", account=account_with_decimals)
- merged = trading_signals_repository._merge_trading_signal_documents(
- _payload_dict(signal),
- {},
- )
- _assert_no_decimal_leaves(merged)
- json.dumps(merged)
- usdc_row = merged["signals"][0]["account"]["copied_assets"][0]
- assert usdc_row["name"] == "USDC"
- assert usdc_row["total"] == 1000.5
- assert usdc_row["available"] == 1000.5
diff --git a/packages/node/tests/functional_tests/test_emit_and_copy_grid_automation_signals.py b/packages/node/tests/functional_tests/test_emit_and_copy_grid_automation_signals.py
index ec4fa29351..17dce2c104 100644
--- a/packages/node/tests/functional_tests/test_emit_and_copy_grid_automation_signals.py
+++ b/packages/node/tests/functional_tests/test_emit_and_copy_grid_automation_signals.py
@@ -12,8 +12,6 @@
#
# You should have received a copy of the GNU General Public License along with
# OctoBot. If not, see https://www.gnu.org/licenses/.
-from __future__ import annotations
-
import asyncio
import datetime
import decimal
@@ -21,6 +19,7 @@
import logging
import mock
import os
+import re
import tempfile
import time
import typing
@@ -31,6 +30,8 @@
import octobot_protocol.models as octobot_protocol_models
import starfish_server.config.schema as starfish_server_config_schema
+import starfish_server.storage.filesystem as starfish_filesystem_storage_module
+import starfish_sharing as starfish_sharing_module
from .util import grid_workflow as grid_sim_util
from .util import price_mocks as price_mocks_module
@@ -79,38 +80,87 @@
_FUNCTIONAL_TEST_SYNC_ENCRYPTION_SECRET = "0123456789abcdef" * 4
-def _grid_functional_test_sync_config():
- """Package default sync config plus a collection for ``TradingSignalsRepository`` HTTP paths."""
- base_config = sync_collections_module.DEFAULT_SYNC_CONFIG
- sync_namespace_key = sync_collections_module.constants.SYNC_NAMESPACE
- assert base_config.namespaces is not None
- assert sync_namespace_key in base_config.namespaces
- octobot_namespace = base_config.namespaces[sync_namespace_key]
- trading_signals_collection = sync_collections_module.CollectionConfig(
- name="trading-signals",
- storagePath="products/{strategyId}/{version}/signals",
- readRoles=["public"],
- writeRoles=["public"],
+_DK_NAMESPACE = "dk"
+
+# Mirrors Infra/sync/server/drakkar_sync/apps/dk_spaces/collections.py's collection set
+# (see octobot_sync.artifacts' module docstring).
+_DK_SIGNAL_COLLECTIONS = [
+ sync_collections_module.CollectionConfig(
+ name="spaces",
+ storagePath="user/{identity}/_spaces",
+ readRoles=["self"],
+ writeRoles=["self"],
encryption="none",
- maxBodyBytes=octobot_sync_constants_module.MAX_BODY_SIZE_SIGNAL,
+ maxBodyBytes=octobot_sync_constants_module.MAX_BODY_SIZE_PRIVATE,
+ ),
+ sync_collections_module.CollectionConfig(
+ name="spaceregistry",
+ storagePath="spaces/{spaceId}/_access",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="none",
+ maxBodyBytes=octobot_sync_constants_module.MAX_BODY_SIZE_PRIVATE,
+ ),
+ sync_collections_module.CollectionConfig(
+ name="spacekeyring",
+ storagePath="spaces/{spaceId}/_keyring",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="none",
+ maxBodyBytes=octobot_sync_constants_module.MAX_BODY_SIZE_PRIVATE,
+ ),
+ sync_collections_module.CollectionConfig(
+ name="objindex",
+ storagePath="spaces/{spaceId}/objects/_index",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="none",
+ maxBodyBytes=octobot_sync_constants_module.MAX_BODY_SIZE_PRIVATE,
+ ),
+ sync_collections_module.CollectionConfig(
+ name="artifact-events",
+ storagePath="spaces/{spaceId}/artifact/versions/{version}/events",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="delegated",
appendOnly=starfish_server_config_schema.AppendOnlyConfig(
type="by_timestamp",
- requireAuthorSignature=False,
+ requireAuthorSignature=True,
),
- )
- extended_octobot = sync_collections_module.NamespaceConfig(
- collections=[*octobot_namespace.collections, trading_signals_collection],
- )
+ maxBodyBytes=octobot_sync_constants_module.MAX_BODY_SIZE_SIGNAL,
+ ),
+]
+
+
+def _grid_functional_test_sync_config():
+ """Package default sync config plus a dk namespace for TradingSignalsRepository's
+ artifact-events publish/pull (see octobot_sync.artifacts)."""
+ base_config = sync_collections_module.DEFAULT_SYNC_CONFIG
+ assert base_config.namespaces is not None
+ dk_namespace = sync_collections_module.NamespaceConfig(collections=_DK_SIGNAL_COLLECTIONS)
return base_config.model_copy(
update={
"namespaces": {
**dict(base_config.namespaces),
- sync_namespace_key: extended_octobot,
+ _DK_NAMESPACE: dk_namespace,
}
}
)
+def _grid_functional_test_dk_role_enricher(object_store):
+ """The generic TOFU space-role enricher, matching Infra/sync's make_space_role_enricher."""
+ return starfish_sharing_module.make_registry_role_enricher(
+ object_store,
+ id_param="spaceId",
+ registry_path="spaces/{id}/_access",
+ owner_role="space:owner",
+ member_role="space:member",
+ allow_tofu=True,
+ id_pattern=re.compile(r"^[a-zA-Z0-9_-]+$"),
+ )
+
+
def _post_fill_open_order_shape(buy_count: int, sell_count: int) -> bool:
"""Accept expected post-shock ladder shapes after forced trigger and mirroring."""
return buy_count == 3 and sell_count == 1
@@ -458,12 +508,19 @@ async def test_emit_and_copy_grid_master_forced_trigger_copies_signals(
# ``NamespaceRewriteMiddleware`` is not applied and Starfish paths
# ``/octobot/v1/...`` never match (HTTP 404). Use the package default
# so the ``octobot`` namespace and rewrite are always active.
+ #
+ # A second FilesystemObjectStore on the same SYNC_DATA_DIR backs the dk
+ # role_enricher, so TOFU role reads see what the server just wrote.
+ dk_role_enricher_store = starfish_filesystem_storage_module.FilesystemObjectStore(
+ starfish_filesystem_storage_module.FilesystemStorageOptions(base_dir=sync_data_dir)
+ )
with mock.patch(
"octobot_sync.app.sync.load_sync_config",
return_value=_grid_functional_test_sync_config(),
):
sync_asgi_app = octobot_sync_server_module.build_default_sync_app(
is_allowed_user_id=lambda _address: True,
+ role_enricher=_grid_functional_test_dk_role_enricher(dk_role_enricher_store),
)
# StarfishClient builds URLs as ``{base}/sync/v1/{namespace}/...``.
# Mount sync_asgi_app under /sync to match the SYNC_MOUNT_PATH prefix.
diff --git a/packages/sync/octobot_sync/app.py b/packages/sync/octobot_sync/app.py
index 27ef2e3b19..d3495939af 100644
--- a/packages/sync/octobot_sync/app.py
+++ b/packages/sync/octobot_sync/app.py
@@ -18,7 +18,7 @@
from fastapi import FastAPI
from starfish_server.storage.base import AbstractObjectStore
-from starfish_server.router.route_builder import create_sync_router, SyncRouterOptions
+from starfish_server.router.route_builder import create_sync_router, SyncRouterOptions, RoleEnricher
from starfish_server.config.schema import ConfigEndpointOptions, SyncConfig
from starfish_protocol.plugins import ServerPlugin
from starfish_server.router.cap_resolver import create_cap_cert_role_resolver, CapAuthError
@@ -136,6 +136,7 @@ def create_app(
sync_config: SyncConfig | None = None,
plugins: list[ServerPlugin] | None = None,
external_host: str | None = None,
+ role_enricher: RoleEnricher | None = None,
):
if sync_config is None:
sync_config = sync.load_sync_config(collections_path)
@@ -149,6 +150,7 @@ def create_app(
store=object_store,
config=sync_config,
role_resolver=_build_role_resolver(is_allowed_user_id),
+ role_enricher=role_enricher,
config_endpoint=ConfigEndpointOptions(auth="public"),
plugins=plugins,
)
diff --git a/packages/sync/octobot_sync/artifacts.py b/packages/sync/octobot_sync/artifacts.py
new file mode 100644
index 0000000000..b9dfd5fe35
--- /dev/null
+++ b/packages/sync/octobot_sync/artifacts.py
@@ -0,0 +1,217 @@
+# This file is part of OctoBot Sync (https://github.com/Drakkar-Software/OctoBot)
+# Copyright (c) 2025 Drakkar-Software, All rights reserved.
+#
+# OctoBot is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either
+# version 3.0 of the License, or (at your option) any later version.
+#
+# OctoBot is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public
+# License along with OctoBot. If not, see .
+
+"""Publish/read dk_spaces' generic ``artifact-events`` collection
+(``spaces/{spaceId}/artifact/versions/{version}/events``, delegated encryption,
+required author signature).
+
+Domain-agnostic: no "signal"/"strategy" concept here — callers derive their own
+``space_id`` via ``artifact_space_id(name)`` (see
+``octobot_flow.repositories.community.trading_signals_repository``).
+
+Scope: publish to your own space, plus ``grant_artifact_space_access`` for a
+KNOWN identity — no invite-link/discovery flow yet.
+
+Note: ``artifact_space_id`` is a hash, not a ``starfish_spaces`` registry
+lookup — the registry write role (``cap:write:spaces``) can never be satisfied
+by a plain device cap, so any wallet's first publish would 403 (confirmed
+against the real server).
+
+TODO(copy-trading): still missing — (a) how a follower learns a space's
+``spaceId``/owner ``userId``, (b) the invite-link path for an unknown identity,
+(c) any UI/API surface calling ``grant_artifact_space_access``.
+"""
+
+import hashlib
+import typing
+
+import starfish_identities
+import starfish_keyring
+import starfish_protocol.types
+import starfish_sdk.types
+import starfish_spaces
+
+import octobot_commons.logging as logging
+
+
+def _artifact_events_name(space_id: str, version: str) -> str:
+ return f"spaces/{space_id}/artifact/versions/{version}/events"
+
+
+def artifact_events_pull_path(space_id: str, version: str) -> str:
+ return f"/pull/{_artifact_events_name(space_id, version)}"
+
+
+def artifact_events_push_path(space_id: str, version: str) -> str:
+ return f"/push/{_artifact_events_name(space_id, version)}"
+
+
+def _logger() -> logging.BotLogger:
+ return logging.get_logger("SyncArtifacts")
+
+
+def artifact_space_id(name: str) -> str:
+ """Deterministic space id for ``name`` — pure, no network I/O (see module docstring)."""
+ digest = hashlib.sha256(name.encode("utf-8")).hexdigest()
+ return f"sp-{digest[:32]}"
+
+
+async def ensure_artifact_space(session: starfish_spaces.Session, space_id: str) -> bool:
+ """Ensure ``spaces/{space_id}/_access`` exists, TOFU-owned by ``session``.
+
+ Returns True if this call created it, False if it already existed. CAS-retried
+ so a concurrent first publish for the same space_id doesn't raise ``ConflictError``.
+ """
+ created = False
+
+ async def _attempt() -> None:
+ nonlocal created
+ entry = await starfish_spaces.read_space_access(session.content_client, space_id, session)
+ if entry.owner is not None:
+ created = False
+ return
+ await starfish_spaces.write_space_access(
+ session.content_client, space_id, session.user_id, [], entry.hash, session, {}
+ )
+ created = True
+
+ await starfish_spaces.run_cas(_attempt)
+ return created
+
+
+async def ensure_artifact_keyring(session: starfish_spaces.Session, space_id: str) -> None:
+ """Create ``space_id``'s space-wide keyring if it doesn't exist yet (owner-only, TOFU)."""
+ await starfish_spaces.owner_ensure_space_keyring(
+ session.content_client,
+ session.keys,
+ space_id,
+ session.layout,
+ starfish_spaces.owner_trusted_adders(session),
+ )
+
+
+async def grant_artifact_space_access(
+ session: starfish_spaces.Session, space_id: str, member_user_id: str, member_kem_pub: str
+) -> None:
+ """Owner-only: add a known identity to the space roster and keyring (no invite link).
+
+ The member reads with their own full session, so this never touches the invite-link
+ cap-scope gap noted in the module docstring.
+ """
+ await starfish_spaces.add_space_member(
+ session.content_client, space_id, session.user_id, member_user_id, session
+ )
+ await starfish_spaces.ensure_space_keyring_recipient(
+ session.content_client,
+ session.keys,
+ space_id,
+ {"subKem": member_kem_pub, "userId": member_user_id},
+ session.layout,
+ starfish_spaces.owner_trusted_adders(session),
+ )
+
+
+async def open_artifact_encryptor(
+ session: starfish_spaces.Session,
+ space_id: str,
+ *,
+ owner_ed_pub: typing.Optional[str] = None,
+) -> typing.Optional[typing.Any]:
+ """Open the delegated encryptor for ``space_id``'s space-wide keyring.
+
+ Returns None ONLY when nothing has been published yet (call ``ensure_artifact_keyring``
+ first when publishing) — a missing keyring pulls as HTTP 200 with an empty body, not a
+ 404. Any other failure (403 not-a-member, 5xx, not-a-recipient) propagates: a caller
+ without access must see a real error, not a silent empty read. ``owner_ed_pub`` defaults
+ to this session's own identity; a copier must pass the owner's ``session.owner_ed_pub``
+ or the trusted-adder set is wrong and every keyring entry gets skipped.
+
+ Pulls/parses the keyring doc directly rather than using
+ ``starfish_spaces.open_encryptor``, which hardcodes the recipient to None as of
+ starfish-spaces==3.0.0a72 and so never finds a wrapped key.
+ """
+ layout = session.layout
+ resolved_owner_ed_pub = owner_ed_pub if owner_ed_pub is not None else session.owner_ed_pub
+ trusted_adders = starfish_identities.compute_owner_trusted_adders(
+ resolved_owner_ed_pub, session.keys["edPub"]
+ )
+ try:
+ result = await session.content_client.pull(layout.keyring_pull(space_id))
+ except starfish_sdk.types.StarfishHttpError as pull_error:
+ if pull_error.status == 404:
+ return None
+ raise
+ if not result.data:
+ _logger().debug(f"No artifact keyring published yet for space {space_id!r}")
+ return None
+ keyring = starfish_keyring.Keyring.from_dict(result.data)
+ return starfish_keyring.create_keyring_encryptor(
+ keyring, session.keys["kemPub"], session.keys["kemPriv"], trusted_adders=trusted_adders
+ )
+
+
+def seal_artifact_payload(encryptor: typing.Any, payload: dict[str, typing.Any]) -> dict[str, typing.Any]:
+ return encryptor.encrypt(payload)
+
+
+def unseal_artifact_payload(encryptor: typing.Any, item: dict[str, typing.Any]) -> dict[str, typing.Any]:
+ return encryptor.decrypt(item["data"])
+
+
+async def publish_artifact_event(
+ session: starfish_spaces.Session,
+ space_id: str,
+ version: str,
+ payload: dict[str, typing.Any],
+ ts: typing.Optional[int] = None,
+) -> starfish_protocol.types.PushSuccess:
+ """Seal ``payload`` under ``space_id``'s keyring and append it to artifact-events."""
+ await ensure_artifact_space(session, space_id)
+ await ensure_artifact_keyring(session, space_id)
+ encryptor = await open_artifact_encryptor(session, space_id)
+ if encryptor is None:
+ raise starfish_sdk.types.StarfishHttpError(
+ 0, f"Could not open the artifact keyring encryptor for space {space_id!r}"
+ )
+ sealed = seal_artifact_payload(encryptor, payload)
+ return await session.content_client.append(
+ artifact_events_push_path(space_id, version), sealed, ts=ts
+ )
+
+
+async def pull_artifact_events(
+ session: starfish_spaces.Session,
+ space_id: str,
+ version: str,
+ last: int,
+ *,
+ owner_ed_pub: typing.Optional[str] = None,
+) -> list[dict[str, typing.Any]]:
+ """Pull and unseal up to ``last`` artifact-event elements for ``space_id``.
+
+ ``owner_ed_pub`` — see ``open_artifact_encryptor``; required for a copier reading
+ someone else's space. Returns [] only when nothing is published yet; any other
+ failure (no access, transport error) propagates.
+ """
+ encryptor = await open_artifact_encryptor(session, space_id, owner_ed_pub=owner_ed_pub)
+ if encryptor is None:
+ return []
+ items = await session.content_client.pull(
+ artifact_events_pull_path(space_id, version), last=last
+ )
+ if not isinstance(items, list):
+ return []
+ return [unseal_artifact_payload(encryptor, item) for item in items]
diff --git a/packages/sync/octobot_sync/enums.py b/packages/sync/octobot_sync/enums.py
index eced01d441..5243ccd9c6 100644
--- a/packages/sync/octobot_sync/enums.py
+++ b/packages/sync/octobot_sync/enums.py
@@ -28,15 +28,5 @@ class Collections(enum.StrEnum):
DEBUG = "debug"
class TemporaryCollections(enum.StrEnum):
- # --- TEMPORARY: append-only product signals collection ---------------------
- # Scaffolding to store signals as an append-only (by_timestamp) log, keyed by
- # PRODUCT (not user identity): every push appends the payload as a {ts, data}
- # element rather than overwriting, and pulls fetch only newer elements via
- # ?checkpoint=. The path is product-scoped, so it carries no {identity} segment
- # and cannot use the "self" role; access is granted to the node's self-signed
- # root device cap (ROLE_ROOT_DEVICE). This whole block (the constant and the
- # CollectionConfig entry below) is temporary and will be REMOVED once the signals
- # storage design is finalized.
- TEMP_PRODUCT_SIGNALS = "product-signals"
# Temporary user-strategies collection; remove when strategies storage is finalized.
TEMP_USER_STRATEGIES = "user-strategies"
\ No newline at end of file
diff --git a/packages/sync/octobot_sync/server.py b/packages/sync/octobot_sync/server.py
index 48a41f2534..154cb4cfb7 100644
--- a/packages/sync/octobot_sync/server.py
+++ b/packages/sync/octobot_sync/server.py
@@ -20,6 +20,7 @@
from collections.abc import Awaitable, Callable
from starfish_server.config.schema import SyncConfig
+from starfish_server.router.route_builder import RoleEnricher
from starfish_server.protocol.types import DOCUMENT_VERSION
from starfish_server.storage.base import AbstractObjectStore, StoreContext
from starfish_server.storage.s3 import S3ObjectStore, S3StorageOptions
@@ -217,10 +218,6 @@ async def get_data(key: str, context: StoreContext | None = None) -> str | None:
_get_identity(context)
)
plaintext = debug_state.to_json()
- case enums.TemporaryCollections.TEMP_PRODUCT_SIGNALS.value:
- # Append-only plaintext log: StoredDocument.data is a dict (items
- # array), not wallet ciphertext — return the persisted doc as-is.
- return await _get_opaque_store().get_string(key)
case _:
# Opaque storage: collections with no protocol bridge are persisted
# as client-encrypted ciphertext and the node never decrypts them.
@@ -240,18 +237,11 @@ async def get_data(key: str, context: StoreContext | None = None) -> str | None:
return _wrap_as_stored_document(encrypted, plaintext)
async def put_data(key: str, body: str, context: StoreContext | None = None) -> None:
- collection = _get_collection(context)
- match collection:
- case enums.TemporaryCollections.TEMP_PRODUCT_SIGNALS.value:
- # Append-only plaintext log: persist the full StoredDocument body
- # (data is a dict), not an unwrapped ciphertext string.
- await _get_opaque_store().put(key, body, content_type="application/json")
- case _:
- # Opaque storage: persist the client ciphertext as-is. The node
- # never decrypts these collections — wallet-key decryption happens
- # entirely on the client.
- ciphertext = _unwrap_stored_document_data(body)
- await _get_opaque_store().put(key, ciphertext, content_type="application/json")
+ # Opaque storage: persist the client ciphertext as-is. The node never
+ # decrypts these collections — wallet-key decryption happens entirely on
+ # the client.
+ ciphertext = _unwrap_stored_document_data(body)
+ await _get_opaque_store().put(key, ciphertext, content_type="application/json")
def set_data_callbacks(
get_data: Callable[[str, StoreContext | None], Awaitable[str | None]],
@@ -313,6 +303,7 @@ def build_default_sync_app(
is_allowed_user_id: Callable[[str], bool] | None = None,
sync_config: SyncConfig | None = None,
external_host: str | None = None,
+ role_enricher: RoleEnricher | None = None,
):
return sync_app.create_app(
build_object_store(),
@@ -320,4 +311,5 @@ def build_default_sync_app(
sync_config=sync_config,
plugins=[user_actions_plugin],
external_host=external_host,
+ role_enricher=role_enricher,
)
diff --git a/packages/sync/octobot_sync/sync/collections.py b/packages/sync/octobot_sync/sync/collections.py
index a42b911b21..c9d5c1fb19 100644
--- a/packages/sync/octobot_sync/sync/collections.py
+++ b/packages/sync/octobot_sync/sync/collections.py
@@ -26,7 +26,6 @@
NamespaceConfig,
AppendOnlyConfig,
)
-from starfish_server.constants import ROLE_ROOT_DEVICE
import octobot_sync.constants as constants
import octobot_sync.enums as enums
@@ -108,25 +107,6 @@
encryption="delegated",
maxBodyBytes=constants.MAX_BODY_SIZE_PRIVATE,
),
- # TEMPORARY (see TemporaryCollections.TEMP_PRODUCT_SIGNALS) —
- # product-scoped append-only signals log. by_timestamp: each push
- # is stored as a {ts, data} element under the "items" array; pulls
- # filter by ?checkpoint=. Keyed by product (productId + version),
- # not user identity. requireAuthorSignature is disabled so the
- # existing (cap-authenticated, but non-author-signing) push path
- # can write without per-element author-proof plumbing.
- CollectionConfig(
- name=enums.TemporaryCollections.TEMP_PRODUCT_SIGNALS.value,
- storagePath="products/{product_id}/{version}/signals",
- readRoles=[ROLE_ROOT_DEVICE],
- writeRoles=[ROLE_ROOT_DEVICE],
- encryption="none",
- maxBodyBytes=constants.MAX_BODY_SIZE_SIGNAL,
- appendOnly=AppendOnlyConfig(
- type="by_timestamp",
- requireAuthorSignature=False,
- ),
- ),
# TEMPORARY (see TemporaryCollections.TEMP_USER_STRATEGIES) —
# temporary user-strategies collection; remove when strategies storage
# is finalized.
diff --git a/packages/sync/tests/artifacts_check.py b/packages/sync/tests/artifacts_check.py
new file mode 100644
index 0000000000..c28d647933
--- /dev/null
+++ b/packages/sync/tests/artifacts_check.py
@@ -0,0 +1,212 @@
+# This file is part of OctoBot Sync (https://github.com/Drakkar-Software/OctoBot)
+# Copyright (c) 2025 Drakkar-Software, All rights reserved.
+#
+# OctoBot is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either
+# version 3.0 of the License, or (at your option) any later version.
+#
+# OctoBot is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public
+# License along with OctoBot. If not, see .
+
+"""Standalone, real-network verification of the dk_spaces artifact publish/pull path.
+
+Not a pytest test — a manual diagnostic script. Walks every step of
+octobot_sync.artifacts against a real (or test) dk_spaces sync server and prints a
+PASS/FAIL line per step.
+
+Usage:
+ python packages/sync/tests/artifacts_check.py \\
+ --private-key 0x... \\
+ --sync-url https://beta-sync.drakkar.software
+
+Use a throwaway/test private key — never a funded wallet's key on the command
+line. --sync-url must be a bare origin (no /sync suffix): build_mirror_session
+appends octobot_sync.constants.SYNC_MOUNT_PATH itself, same as every other
+caller in this codebase (see octobot_sync/client.py, octobot/constants.py).
+"""
+
+import argparse
+import asyncio
+import sys
+import time
+import typing
+
+import octobot_sync.mirror.writer as mirror_writer
+import octobot_sync.artifacts as artifacts
+
+
+def _pass(step: str, detail: str = "") -> None:
+ print(f"[PASS] {step}" + (f" — {detail}" if detail else ""))
+
+
+def _fail(step: str, detail: str) -> None:
+ print(f"[FAIL] {step} — {detail}")
+
+
+async def run_check(private_key: str, sync_url: str, artifact_name: str, version: str) -> bool:
+ """Run every step in sequence; stop and return False at the first failure."""
+ payload = {"artifacts_check": True, "sent_at": time.time()}
+
+ try:
+ derived = mirror_writer.derived_identity_for_mirror(private_key)
+ except Exception as err:
+ _fail("derive identity", str(err))
+ return False
+ _pass("derive identity", f"user_id={derived['userId']}")
+
+ try:
+ session = await mirror_writer.build_mirror_session(derived, sync_url, name="artifacts-check")
+ except Exception as err:
+ _fail("build session", str(err))
+ return False
+ _pass("build session", f"namespace={session.namespace} sync_url={sync_url}")
+
+ space_id = artifacts.artifact_space_id(artifact_name)
+ try:
+ created = await artifacts.ensure_artifact_space(session, space_id)
+ except Exception as err:
+ _fail("artifact space creation", str(err))
+ return False
+ _pass(
+ "artifact space creation",
+ f"{'created new' if created else 'reused existing'} space_id={space_id}",
+ )
+
+ try:
+ await artifacts.ensure_artifact_keyring(session, space_id)
+ except Exception as err:
+ _fail("ensure keyring", str(err))
+ return False
+ _pass("ensure keyring")
+
+ try:
+ publish_encryptor = await artifacts.open_artifact_encryptor(session, space_id)
+ except Exception as err:
+ _fail("open encryptor (publish)", str(err))
+ return False
+ if publish_encryptor is None:
+ _fail("open encryptor (publish)", "returned None — keyring not accessible")
+ return False
+ _pass("open encryptor (publish)")
+
+ try:
+ sealed = artifacts.seal_artifact_payload(publish_encryptor, payload)
+ except Exception as err:
+ _fail("seal payload", str(err))
+ return False
+ _pass("seal payload", f"{len(sealed.get('_encrypted', ''))} base64 chars")
+
+ try:
+ ts = int(payload["sent_at"] * 1000)
+ push_result = await session.content_client.append(
+ artifacts.artifact_events_push_path(space_id, version), sealed, ts=ts
+ )
+ except Exception as err:
+ _fail("publish (append)", str(err))
+ return False
+ _pass("publish (append)", f"hash={push_result.hash} ts={push_result.timestamp}")
+
+ try:
+ # A real server round-trip: confirms the space persisted server-side.
+ created_again = await artifacts.ensure_artifact_space(session, space_id)
+ except Exception as err:
+ _fail("resolve space (pull)", str(err))
+ return False
+ if created_again:
+ _fail("resolve space (pull)", "space_access was missing server-side and got re-created")
+ return False
+ _pass("resolve space (pull)", f"space_id={space_id} confirmed to exist server-side")
+
+ try:
+ pull_encryptor = await artifacts.open_artifact_encryptor(session, space_id)
+ except Exception as err:
+ _fail("open encryptor (pull)", str(err))
+ return False
+ if pull_encryptor is None:
+ _fail("open encryptor (pull)", "returned None — keyring not accessible")
+ return False
+ _pass("open encryptor (pull)")
+
+ try:
+ items = await session.content_client.pull(
+ artifacts.artifact_events_pull_path(space_id, version), last=10
+ )
+ except Exception as err:
+ _fail("event fetching (pull)", str(err))
+ return False
+ if not isinstance(items, list):
+ _fail("event fetching (pull)", f"unexpected response type {type(items)}")
+ return False
+ _pass("event fetching (pull)", f"{len(items)} item(s)")
+
+ try:
+ unsealed = [artifacts.unseal_artifact_payload(pull_encryptor, item) for item in items]
+ except Exception as err:
+ _fail("unseal events", str(err))
+ return False
+ if payload not in unsealed:
+ _fail("unseal + round-trip", "published payload not found among pulled/unsealed items")
+ return False
+ _pass("unseal + round-trip", "published payload found intact")
+
+ return True
+
+
+def _build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ description=(
+ "Real-network check of the dk_spaces artifact publish/pull path (artifact space "
+ "creation, keyring setup, sealing, publish/append, event fetching) against a "
+ "dk_spaces sync server. Prints a PASS/FAIL line per step."
+ )
+ )
+ parser.add_argument(
+ "--private-key",
+ required=True,
+ help="EVM private key (0x-prefixed hex) to derive the test wallet identity from — "
+ "use a throwaway/test key, never a funded one",
+ )
+ parser.add_argument(
+ "--sync-url",
+ required=True,
+ help="Bare sync server origin, e.g. https://beta-sync.drakkar.software "
+ "(no /sync suffix — build_mirror_session appends it)",
+ )
+ parser.add_argument(
+ "--artifact-name",
+ default=f"artifacts-check-{int(time.time())}",
+ help="Artifact space name to publish/pull under (default: a fresh generated name so "
+ "repeat runs don't collide)",
+ )
+ parser.add_argument(
+ "--version",
+ default="1.0.0",
+ help="artifact-events version segment (default: 1.0.0)",
+ )
+ return parser
+
+
+async def async_main(argv: typing.Optional[list[str]] = None) -> int:
+ """The async core of main(), split out so a caller already inside a running event loop
+ (e.g. a pytest-asyncio test) can await it directly instead of nesting an asyncio.run()."""
+ args = _build_parser().parse_args(argv)
+
+ print(f"Checking artifact publishing for name={args.artifact_name!r} against {args.sync_url}\n")
+ ok = await run_check(args.private_key, args.sync_url, args.artifact_name, args.version)
+ print()
+ print("RESULT: all steps passed" if ok else "RESULT: FAILED — see the first [FAIL] line above")
+ return 0 if ok else 1
+
+
+def main(argv: typing.Optional[list[str]] = None) -> int:
+ return asyncio.run(async_main(argv))
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/packages/sync/tests/test_artifacts_check.py b/packages/sync/tests/test_artifacts_check.py
new file mode 100644
index 0000000000..5eb20579ba
--- /dev/null
+++ b/packages/sync/tests/test_artifacts_check.py
@@ -0,0 +1,130 @@
+# This file is part of OctoBot Sync (https://github.com/Drakkar-Software/OctoBot)
+# Copyright (c) 2025 Drakkar-Software, All rights reserved.
+#
+# OctoBot is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either
+# version 3.0 of the License, or (at your option) any later version.
+#
+# OctoBot is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public
+# License along with OctoBot. If not, see .
+
+"""Tests for tests/artifacts_check.py's step composition and PASS/FAIL reporting.
+
+artifacts_check.run_check builds its own session from a raw sync_url, so — unlike
+test_integration_artifacts.py's ASGITransport fixture — this needs a real local
+uvicorn server serving the same dk_app fixture.
+"""
+
+import asyncio
+
+import pytest
+import uvicorn
+
+import octobot_commons.os_util as commons_os_util
+
+import tests.test_integration_artifacts as integration_fixtures
+import tests.artifacts_check as artifacts_check
+
+
+# Well-known Hardhat test key #2 — public, safe to embed in tests.
+_PRIV = "0x5de4111afa1a4b94908f83103eb1f1706367c2e68ca870fc3fb9a804cdab365"
+
+
+@pytest.fixture
+def store():
+ return integration_fixtures.MemoryObjectStore()
+
+
+@pytest.fixture
+async def sync_url(store):
+ """A real uvicorn server serving the same dk_app config as test_integration_artifacts.py."""
+ app = integration_fixtures.sync_app.create_app(
+ store,
+ sync_config=integration_fixtures._DK_CONFIG,
+ role_enricher=integration_fixtures._make_dk_role_enricher(store),
+ )
+ port = commons_os_util.find_first_free_listen_port_after_base("127.0.0.1", 31500, max_offset=256)
+ server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning"))
+ serve_task = asyncio.create_task(server.serve())
+ await asyncio.sleep(0.25)
+ try:
+ yield f"http://127.0.0.1:{port}"
+ finally:
+ server.should_exit = True
+ await serve_task
+
+
+class TestRunCheck:
+ @pytest.mark.asyncio
+ async def test_all_steps_pass_and_report_correctly(self, sync_url, capsys):
+ ok = await artifacts_check.run_check(_PRIV, sync_url, "check-artifact-1", "1.0.0")
+
+ assert ok is True
+ out = capsys.readouterr().out
+ assert "[FAIL]" not in out
+ for step in (
+ "derive identity",
+ "build session",
+ "artifact space creation",
+ "ensure keyring",
+ "open encryptor (publish)",
+ "seal payload",
+ "publish (append)",
+ "resolve space (pull)",
+ "open encryptor (pull)",
+ "event fetching (pull)",
+ "unseal + round-trip",
+ ):
+ assert f"[PASS] {step}" in out
+
+ @pytest.mark.asyncio
+ async def test_reports_created_new_then_reused_existing(self, sync_url, capsys):
+ await artifacts_check.run_check(_PRIV, sync_url, "check-artifact-2", "1.0.0")
+ first_out = capsys.readouterr().out
+ assert "created new" in first_out
+
+ await artifacts_check.run_check(_PRIV, sync_url, "check-artifact-2", "1.0.0")
+ second_out = capsys.readouterr().out
+ assert "reused existing" in second_out
+
+ @pytest.mark.asyncio
+ async def test_fails_cleanly_on_an_unreachable_server(self, capsys):
+ ok = await artifacts_check.run_check(
+ _PRIV, "http://127.0.0.1:1", "check-artifact-3", "1.0.0"
+ )
+
+ assert ok is False
+ out = capsys.readouterr().out
+ assert "[FAIL]" in out
+ assert "RESULT" not in out
+
+
+class TestMain:
+ @pytest.mark.asyncio
+ async def test_returns_zero_on_success(self, sync_url, capsys):
+ # async_main directly, not main(): main()'s own asyncio.run() would leave
+ # sync_url's background uvicorn task (on this test's event loop) unpumped mid-test.
+ exit_code = await artifacts_check.async_main(
+ ["--private-key", _PRIV, "--sync-url", sync_url, "--artifact-name", "check-artifact-4"]
+ )
+
+ assert exit_code == 0
+ assert "RESULT: all steps passed" in capsys.readouterr().out
+
+ def test_returns_nonzero_on_failure(self, capsys):
+ exit_code = artifacts_check.main(
+ [
+ "--private-key", _PRIV,
+ "--sync-url", "http://127.0.0.1:1",
+ "--artifact-name", "check-artifact-5",
+ ]
+ )
+
+ assert exit_code == 1
+ assert "RESULT: FAILED" in capsys.readouterr().out
diff --git a/packages/sync/tests/test_client.py b/packages/sync/tests/test_client.py
new file mode 100644
index 0000000000..bfda701d46
--- /dev/null
+++ b/packages/sync/tests/test_client.py
@@ -0,0 +1,49 @@
+# Drakkar-Software OctoBot-Sync
+# Copyright (c) Drakkar-Software, All rights reserved.
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 3.0 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library.
+
+"""Tests for create_sync_client's base_url composition.
+
+Guards against the mount-path doubling regression: create_sync_client appends
+SYNC_MOUNT_PATH itself, so callers MUST pass a bare origin.
+"""
+
+import octobot_sync.client as sync_client
+import octobot_sync.constants as constants
+
+# Well-known Hardhat test key #1 — public, safe to embed in tests.
+_TEST_PRIVATE_KEY = "0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d"
+
+
+def test_create_sync_client_appends_mount_path_once_to_a_bare_origin():
+ client, _user_id = sync_client.create_sync_client(
+ _TEST_PRIVATE_KEY, sync_url="https://prod-sync.drakkar.software"
+ )
+ assert client._base_url == f"https://prod-sync.drakkar.software/{constants.SYNC_MOUNT_PATH}"
+
+
+def test_create_sync_client_strips_a_trailing_slash_before_appending_mount_path():
+ client, _user_id = sync_client.create_sync_client(
+ _TEST_PRIVATE_KEY, sync_url="https://prod-sync.drakkar.software/"
+ )
+ assert client._base_url == f"https://prod-sync.drakkar.software/{constants.SYNC_MOUNT_PATH}"
+
+
+def test_create_sync_client_does_not_double_an_already_suffixed_origin():
+ # Documents the actual (buggy) behavior for an already-suffixed sync_url.
+ client, _user_id = sync_client.create_sync_client(
+ _TEST_PRIVATE_KEY, sync_url="https://prod-sync.drakkar.software/sync"
+ )
+ assert client._base_url == "https://prod-sync.drakkar.software/sync/sync"
diff --git a/packages/sync/tests/test_integration_artifacts.py b/packages/sync/tests/test_integration_artifacts.py
new file mode 100644
index 0000000000..5f1bd8b2ce
--- /dev/null
+++ b/packages/sync/tests/test_integration_artifacts.py
@@ -0,0 +1,443 @@
+# This file is part of OctoBot Sync (https://github.com/Drakkar-Software/OctoBot)
+# Copyright (c) 2025 Drakkar-Software, All rights reserved.
+#
+# OctoBot is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either
+# version 3.0 of the License, or (at your option) any later version.
+#
+# OctoBot is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public
+# License along with OctoBot. If not, see .
+
+"""End-to-end integration of octobot_sync.artifacts against a real in-process dk
+namespace: a real starfish_spaces Session, a real TOFU role enricher
+(starfish_sharing.make_registry_role_enricher), and a real starfish_server app
+mounted under /sync.
+"""
+
+import asyncio
+import re
+from unittest import mock
+
+import httpx
+import pytest
+from fastapi import FastAPI
+
+from starfish_sdk import StarfishClient
+import starfish_sdk.types
+from starfish_server.config.schema import (
+ SyncConfig,
+ CollectionConfig,
+ NamespaceConfig,
+ AppendOnlyConfig,
+)
+from starfish_sharing import make_registry_role_enricher
+from starfish_spaces.client import ClientOpts, DeviceKeys
+from starfish_spaces.session import BuildSessionOpts, Session, build_session
+
+import octobot_sync.app as sync_app
+import octobot_sync.constants as constants
+import octobot_sync.artifacts as artifacts
+
+
+class MemoryObjectStore:
+ """Minimal in-memory AbstractObjectStore (every method takes *, context=None)."""
+
+ def __init__(self) -> None:
+ self._store: dict[str, str] = {}
+
+ async def get_string(self, key, *, context=None):
+ return self._store.get(key)
+
+ async def put(self, key, body, *, content_type=None, cache_control=None, context=None):
+ self._store[key] = body
+
+ async def list_keys(self, prefix, *, start_after=None, limit=None, context=None):
+ keys = sorted(k for k in self._store if k.startswith(prefix))
+ if start_after:
+ keys = [k for k in keys if k > start_after]
+ return keys[:limit] if limit else keys
+
+ async def delete(self, key, *, context=None):
+ self._store.pop(key, None)
+
+ async def delete_many(self, keys, *, context=None):
+ for k in keys:
+ self._store.pop(k, None)
+
+
+_DK_NAMESPACE = "dk"
+
+_DK_CONFIG = SyncConfig(
+ version=1,
+ collections=[],
+ namespaces={
+ _DK_NAMESPACE: NamespaceConfig(
+ collections=[
+ # No "spaces" or "objindex" collection: artifacts.py never touches either
+ # (see its module docstring). maxBodyBytes on these two matches the real
+ # deployed values (131_072 / 65_536), not MAX_BODY_SIZE_PRIVATE.
+ CollectionConfig(
+ name="spaceregistry",
+ storagePath="spaces/{spaceId}/_access",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="none",
+ maxBodyBytes=131_072,
+ ),
+ CollectionConfig(
+ name="spacekeyring",
+ storagePath="spaces/{spaceId}/_keyring",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="none",
+ maxBodyBytes=65_536,
+ ),
+ # Mirrors Infra/sync/server/drakkar_sync/apps/dk_spaces/collections.py's
+ # "artifact-events" collection exactly.
+ CollectionConfig(
+ name="artifact-events",
+ storagePath="spaces/{spaceId}/artifact/versions/{version}/events",
+ readRoles=["space:member"],
+ writeRoles=["space:owner"],
+ encryption="delegated",
+ appendOnly=AppendOnlyConfig(type="by_timestamp", requireAuthorSignature=True),
+ maxBodyBytes=constants.MAX_BODY_SIZE_SIGNAL,
+ ),
+ ]
+ ),
+ },
+)
+
+
+def _make_dk_role_enricher(store: MemoryObjectStore):
+ return make_registry_role_enricher(
+ store,
+ id_param="spaceId",
+ registry_path="spaces/{id}/_access",
+ owner_role="space:owner",
+ member_role="space:member",
+ allow_tofu=True,
+ id_pattern=re.compile(r"^[a-zA-Z0-9_-]+$"),
+ )
+
+
+# Well-known Anvil/Hardhat account #1 (0x7099...79C8) — public, safe to embed in tests.
+_PRIV = "0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d"
+# Well-known Anvil/Hardhat account #2 (0x3C44...93BC), matches tests/e2e/conftest.py's
+# OTHER_PRIVATE_KEY — a second, distinct identity used as the copy-trading copier.
+_COPIER_PRIV = "0x5de4111afa1a4b94908f83103eb1f1706367c2e68ca870fc3fb9a804cdab365a"
+
+
+@pytest.fixture
+def store():
+ return MemoryObjectStore()
+
+
+@pytest.fixture
+def dk_app(store):
+ inner = sync_app.create_app(
+ store,
+ sync_config=_DK_CONFIG,
+ role_enricher=_make_dk_role_enricher(store),
+ )
+ outer = FastAPI()
+ outer.mount("/sync", inner) # mount exactly as node_api does
+ return outer
+
+
+async def _build_session(dk_app, private_key: str) -> Session:
+ """A real starfish_spaces Session for `private_key`, wired to the in-process dk_app.
+
+ starfish_spaces.session.build_session builds its own StarfishClients internally with
+ no transport-injection hook, so starfish_spaces.session.make_space_client is
+ monkeypatched for the duration of this call. Callable more than once per test (a
+ fresh Session for the same identity, or a second identity for a copier).
+ """
+ transport = httpx.ASGITransport(app=dk_app)
+
+ from octobot_sync.auth.provider import derive_root_identity
+ import starfish_spaces.client as spaces_client_module
+ import starfish_spaces.session as spaces_session_module
+
+ root = derive_root_identity(private_key)
+ keys: DeviceKeys = {
+ "edPriv": root.keys.ed_priv,
+ "edPub": root.keys.ed_pub,
+ "kemPriv": root.keys.kem_priv,
+ "kemPub": root.keys.kem_pub,
+ }
+ client_opts: ClientOpts = {"baseUrl": "http://test/sync", "namespace": _DK_NAMESPACE}
+
+ def _make_space_client(cap, ed_priv_hex, opts):
+ return StarfishClient(
+ opts["baseUrl"],
+ cap_provider=spaces_client_module.cap_provider_for(cap, ed_priv_hex),
+ namespace=opts.get("namespace"),
+ timeout=float(opts.get("timeout", 30.0)),
+ client=httpx.AsyncClient(transport=transport),
+ )
+
+ original = spaces_session_module.make_space_client
+ spaces_session_module.make_space_client = _make_space_client
+ try:
+ return await build_session(
+ BuildSessionOpts(user_id=root.user_id, keys=keys, client_opts=client_opts)
+ )
+ finally:
+ spaces_session_module.make_space_client = original
+
+
+@pytest.fixture
+async def artifact_session(dk_app):
+ session = await _build_session(dk_app, _PRIV)
+ yield session
+ await session.content_client.close()
+ await session.account_client.close()
+
+
+class TestPublishAndPullArtifactEvent:
+ @pytest.mark.asyncio
+ async def test_round_trips_a_sealed_signed_event(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-1")
+ payload = {"kind": "test", "value": 1000.0}
+
+ push_result = await artifacts.publish_artifact_event(
+ artifact_session, space_id, "1.0.0", payload, ts=1000
+ )
+ assert push_result.hash
+
+ pulled = await artifacts.pull_artifact_events(artifact_session, space_id, "1.0.0", last=10)
+ assert pulled == [payload]
+
+ @pytest.mark.asyncio
+ async def test_two_events_are_returned_in_append_order(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-2")
+ first = {"kind": "test", "value": 1000.0}
+ second = {"kind": "test", "value": 2000.0}
+
+ await artifacts.publish_artifact_event(artifact_session, space_id, "1.0.0", first, ts=1000)
+ await artifacts.publish_artifact_event(artifact_session, space_id, "1.0.0", second, ts=2000)
+
+ pulled = await artifacts.pull_artifact_events(artifact_session, space_id, "1.0.0", last=10)
+ assert pulled == [first, second]
+
+ @pytest.mark.asyncio
+ async def test_pull_of_an_unknown_space_returns_empty(self, artifact_session):
+ space_id = artifacts.artifact_space_id("no-such-space")
+ pulled = await artifacts.pull_artifact_events(artifact_session, space_id, "1.0.0", last=10)
+ assert pulled == []
+
+ @pytest.mark.asyncio
+ async def test_reuses_the_same_space_across_publishes(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-3")
+
+ await artifacts.publish_artifact_event(artifact_session, space_id, "1.0.0", {"a": 1}, ts=1000)
+ # Real server round-trip: confirms the space's _access doc persisted server-side.
+ created_again = await artifacts.ensure_artifact_space(artifact_session, space_id)
+ assert created_again is False
+
+ @pytest.mark.asyncio
+ async def test_ensure_artifact_space_reports_created_then_reused(self, artifact_session):
+ # Regression test: a missing JSON doc pulls as 200 OK with an empty dict, not a
+ # 404/exception. ensure_artifact_space originally treated "the pull didn't raise" as
+ # "already exists" — confirmed against the real deployed dk_spaces server. Must use
+ # a never-touched space id to catch this.
+ space_id = artifacts.artifact_space_id("brand-new-never-published-space")
+
+ created_first = await artifacts.ensure_artifact_space(artifact_session, space_id)
+ assert created_first is True
+
+ created_second = await artifacts.ensure_artifact_space(artifact_session, space_id)
+ assert created_second is False
+
+ @pytest.mark.asyncio
+ async def test_a_fresh_session_for_the_same_identity_can_read_back_published_events(
+ self, dk_app, artifact_session
+ ):
+ # artifact_session and this freshly-built one are two INDEPENDENT Session objects for
+ # the same identity (_PRIV), simulating a bot restart. Proves decryption doesn't
+ # depend on in-process state — it must be re-derivable from the private key plus a
+ # fresh keyring pull.
+ space_id = artifacts.artifact_space_id("space-fresh")
+ payload = {"kind": "test", "value": 1000.0}
+ await artifacts.publish_artifact_event(artifact_session, space_id, "1.0.0", payload, ts=1000)
+
+ fresh_session = await _build_session(dk_app, _PRIV)
+ try:
+ pulled = await artifacts.pull_artifact_events(fresh_session, space_id, "1.0.0", last=10)
+ finally:
+ await fresh_session.content_client.close()
+ await fresh_session.account_client.close()
+
+ assert pulled == [payload]
+
+
+class TestCopyTrading:
+ """Publisher (_PRIV) grants a copier (_COPIER_PRIV, its own independent Session) access
+ via grant_artifact_space_access — the known-identity path, not an invite link."""
+
+ @pytest.mark.asyncio
+ async def test_copier_read_raises_before_being_granted_access(
+ self, dk_app, artifact_session
+ ):
+ space_id = artifacts.artifact_space_id("copy-space-gate")
+ await artifacts.publish_artifact_event(artifact_session, space_id, "1.0.0", {"a": 1}, ts=1000)
+
+ copier_session = await _build_session(dk_app, _COPIER_PRIV)
+ try:
+ with pytest.raises(starfish_sdk.types.StarfishHttpError) as excinfo:
+ await artifacts.pull_artifact_events(copier_session, space_id, "1.0.0", last=10)
+ finally:
+ await copier_session.content_client.close()
+ await copier_session.account_client.close()
+
+ assert excinfo.value.status == 403
+
+ @pytest.mark.asyncio
+ async def test_copier_reads_the_event_after_being_granted_access(
+ self, dk_app, artifact_session
+ ):
+ space_id = artifacts.artifact_space_id("copy-space-granted")
+ payload = {"kind": "test", "value": 1000.0}
+ await artifacts.publish_artifact_event(artifact_session, space_id, "1.0.0", payload, ts=1000)
+
+ from octobot_sync.auth.provider import derive_root_identity
+
+ copier_root = derive_root_identity(_COPIER_PRIV)
+ copier_session = await _build_session(dk_app, _COPIER_PRIV)
+ try:
+ await artifacts.grant_artifact_space_access(
+ artifact_session, space_id, copier_root.user_id, copier_root.keys.kem_pub
+ )
+
+ # owner_ed_pub is required here: without it, pull_artifact_events would treat
+ # the copier as its own space owner and compute the wrong trusted-adder set.
+ pulled = await artifacts.pull_artifact_events(
+ copier_session, space_id, "1.0.0", last=10, owner_ed_pub=artifact_session.owner_ed_pub
+ )
+ finally:
+ await copier_session.content_client.close()
+ await copier_session.account_client.close()
+
+ assert pulled == [payload]
+
+
+class TestArtifactSpaceId:
+ """Pure, no network — artifact_space_id must never depend on server state."""
+
+ def test_is_deterministic_for_the_same_name(self):
+ assert artifacts.artifact_space_id("name-x") == artifacts.artifact_space_id("name-x")
+
+ def test_differs_across_names(self):
+ assert artifacts.artifact_space_id("name-x") != artifacts.artifact_space_id("name-y")
+
+ def test_matches_the_expected_shape(self):
+ space_id = artifacts.artifact_space_id("name-x")
+ assert space_id.startswith("sp-")
+ digest = space_id.removeprefix("sp-")
+ assert len(digest) == 32
+ assert all(c in "0123456789abcdef" for c in digest)
+
+
+class TestBodySizeLimit:
+ """artifact-events caps at 65_536 bytes, 4x tighter than the old octobot/products path."""
+
+ @pytest.mark.asyncio
+ async def test_a_realistically_sized_payload_publishes_fine(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-size-ok")
+ payload = {
+ "kind": "test",
+ "snapshots": [{"value": i} for i in range(50)],
+ }
+ push_result = await artifacts.publish_artifact_event(
+ artifact_session, space_id, "1.0.0", payload, ts=1000
+ )
+ assert push_result.hash
+
+ @pytest.mark.asyncio
+ async def test_a_payload_over_the_cap_is_rejected_with_a_clear_error(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-too-big")
+ payload = {
+ "kind": "test",
+ "snapshots": [{"value": i, "padding": "x" * 200} for i in range(400)],
+ }
+ with pytest.raises(starfish_sdk.types.StarfishHttpError) as excinfo:
+ await artifacts.publish_artifact_event(artifact_session, space_id, "1.0.0", payload, ts=1000)
+ assert excinfo.value.status in (400, 413)
+
+
+class TestEnsureArtifactKeyring:
+ @pytest.mark.asyncio
+ async def test_a_fresh_keyring_makes_publishing_identity_able_to_decrypt(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-keyring-fresh")
+ await artifacts.ensure_artifact_space(artifact_session, space_id)
+
+ # Nothing published yet: a missing keyring pulls as HTTP 200 + empty body, not a 404.
+ assert await artifacts.open_artifact_encryptor(artifact_session, space_id) is None
+
+ await artifacts.ensure_artifact_keyring(artifact_session, space_id)
+ assert await artifacts.open_artifact_encryptor(artifact_session, space_id) is not None
+
+ @pytest.mark.asyncio
+ async def test_ensuring_an_existing_keyring_twice_is_a_no_op(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-keyring-idempotent")
+ await artifacts.ensure_artifact_space(artifact_session, space_id)
+ await artifacts.ensure_artifact_keyring(artifact_session, space_id)
+
+ # Must not raise, and the keyring must still be usable afterwards.
+ await artifacts.ensure_artifact_keyring(artifact_session, space_id)
+ assert await artifacts.open_artifact_encryptor(artifact_session, space_id) is not None
+
+
+class TestOpenArtifactEncryptorPropagatesRealFailures:
+ """Only a genuinely missing keyring (404, or 200+empty body) is treated as empty —
+ anything else is a real failure the caller must see, not a silent None/[]."""
+
+ @pytest.mark.asyncio
+ async def test_a_non_404_pull_failure_propagates(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-keyring-server-error")
+ await artifacts.ensure_artifact_space(artifact_session, space_id)
+ await artifacts.ensure_artifact_keyring(artifact_session, space_id)
+
+ with mock.patch.object(
+ artifact_session.content_client,
+ "pull",
+ mock.AsyncMock(side_effect=starfish_sdk.types.StarfishHttpError(500, "boom")),
+ ):
+ with pytest.raises(starfish_sdk.types.StarfishHttpError) as excinfo:
+ await artifacts.open_artifact_encryptor(artifact_session, space_id)
+
+ assert excinfo.value.status == 500
+
+
+class TestEnsureArtifactSpaceConcurrency:
+ @pytest.mark.asyncio
+ async def test_two_concurrent_first_publishes_agree_on_a_single_creator(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-cas-race")
+
+ results = await asyncio.gather(
+ artifacts.ensure_artifact_space(artifact_session, space_id),
+ artifacts.ensure_artifact_space(artifact_session, space_id),
+ )
+
+ assert sorted(results) == [False, True]
+
+
+class TestPublishArtifactEventMissingEncryptor:
+ @pytest.mark.asyncio
+ async def test_raises_a_clear_error_when_the_encryptor_cannot_be_opened(self, artifact_session):
+ space_id = artifacts.artifact_space_id("space-no-encryptor")
+
+ with mock.patch.object(artifacts, "open_artifact_encryptor", return_value=None):
+ with pytest.raises(starfish_sdk.types.StarfishHttpError) as excinfo:
+ await artifacts.publish_artifact_event(
+ artifact_session, space_id, "1.0.0", {"a": 1}, ts=1000
+ )
+
+ assert space_id in str(excinfo.value)
diff --git a/packages/sync/tests/test_integration_cap_sync.py b/packages/sync/tests/test_integration_cap_sync.py
index ab33525ff3..d2e988b517 100644
--- a/packages/sync/tests/test_integration_cap_sync.py
+++ b/packages/sync/tests/test_integration_cap_sync.py
@@ -85,12 +85,11 @@ async def delete_many(self, keys, *, context=None):
encryption="none",
maxBodyBytes=constants.MAX_BODY_SIZE_PRIVATE,
),
- # Product-scoped append-only log (mirrors the temporary
- # product-signals collection): no {identity} segment, authorized
+ # Generic append-only log with no {identity} segment, authorized
# by the node's self-signed root device cap (device:root).
CollectionConfig(
- name="product-signals",
- storagePath="products/{product_id}/{version}/signals",
+ name="root-scoped-log",
+ storagePath="feeds/{feed_id}/{version}/events",
readRoles=[ROLE_ROOT_DEVICE],
writeRoles=[ROLE_ROOT_DEVICE],
encryption="none",
@@ -155,11 +154,11 @@ async def test_cap_signed_large_body_not_413():
@pytest.mark.asyncio
-async def test_cap_signed_product_append_only_collection():
- # Product-scoped append-only path (no {identity}) authorized via device:root.
+async def test_cap_signed_root_scoped_append_only_collection():
+ # Root-scoped append-only path (no {identity}) authorized via device:root.
client, http_client, _user_id = _make_client()
try:
- path = "products/prod-123/v1/signals"
+ path = "feeds/feed-123/v1/events"
await client.append(f"/push/{path}", {"n": 1})
await client.append(f"/push/{path}", {"n": 2})
items = await client.pull(f"/pull/{path}", append_field="items", full=True)
diff --git a/packages/sync/tests/test_integration_product_signals_server.py b/packages/sync/tests/test_integration_product_signals_server.py
deleted file mode 100644
index 99f6d8f0df..0000000000
--- a/packages/sync/tests/test_integration_product_signals_server.py
+++ /dev/null
@@ -1,125 +0,0 @@
-# This file is part of OctoBot Sync (https://github.com/Drakkar-Software/OctoBot)
-# Copyright (c) 2025 Drakkar-Software, All rights reserved.
-#
-# OctoBot is free software; you can redistribute it and/or
-# modify it under the terms of the GNU General Public License
-# as published by the Free Software Foundation; either
-# version 3.0 of the License, or (at your option) any later version.
-#
-# OctoBot is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-# General Public License for more details.
-#
-# You should have received a copy of the GNU General Public
-# License along with OctoBot. If not, see .
-
-"""End-to-end product-signals storage via server get_data/put_data callbacks.
-
-Mirrors node_api wiring (set_data_callbacks + build_default_sync_app) so
-append-only plaintext documents with dict ``data`` round-trip through the
-opaque filesystem store.
-"""
-
-import httpx
-import mock
-import pytest
-from fastapi import FastAPI
-
-from starfish_sdk import StarfishClient
-
-import octobot_commons.user_root_folder_provider as user_root_folder_provider
-import octobot_sync.auth as auth
-import octobot_sync.client as sync_client
-import octobot_sync.constants as constants
-import octobot_sync.server as sync_server
-import octobot_sync.sync.collections as sync_collections
-
-# Well-known test key (Anvil account #1).
-_PRIV = "0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d"
-
-_STRATEGY_ID = "fc7982e2-cfbf-41ba-8f5a-38045b27883a"
-_VERSION = "1.0.0"
-
-
-def _signal_payload(updated_at: float) -> dict:
- return {
- "strategy_id": _STRATEGY_ID,
- "account": {"updated_at": updated_at},
- }
-
-
-def _push_path() -> str:
- return f"/v1/push/products/{_STRATEGY_ID}/{_VERSION}/signals"
-
-
-def _pull_path() -> str:
- return f"/v1/pull/products/{_STRATEGY_ID}/{_VERSION}/signals"
-
-
-@pytest.fixture
-def callback_sync_app(tmp_path):
- original_get = sync_server._get_data
- original_put = sync_server._put_data
- original_opaque = sync_server._opaque_store
- sync_server.set_data_callbacks(sync_server.get_data, sync_server.put_data)
- with mock.patch.object(
- user_root_folder_provider,
- "get_user_root_folder",
- return_value=str(tmp_path),
- ):
- inner = sync_server.build_default_sync_app(
- sync_config=sync_collections.DEFAULT_SYNC_CONFIG,
- )
- root = FastAPI()
- root.mount("/sync", inner)
- yield root
- sync_server._get_data = original_get
- sync_server._put_data = original_put
- sync_server._opaque_store = original_opaque
-
-
-@pytest.fixture
-async def starfish_http_client(callback_sync_app):
- http_client = httpx.AsyncClient(
- transport=httpx.ASGITransport(app=callback_sync_app),
- base_url="http://test/sync",
- )
- yield http_client
- await http_client.aclose()
-
-
-@pytest.fixture
-def starfish_client(starfish_http_client):
- cap_provider = auth.WalletCapProvider(_PRIV)
- return StarfishClient(
- base_url="http://test/sync",
- cap_provider=cap_provider,
- namespace=constants.SYNC_NAMESPACE,
- client=starfish_http_client,
- )
-
-
-@pytest.mark.asyncio
-async def test_append_signals_roundtrip_via_server_callbacks(starfish_client):
- first_signal = _signal_payload(1.0)
- second_signal = _signal_payload(2.0)
- await sync_client.append_payload(
- starfish_client,
- push_path=_push_path(),
- payload=first_signal,
- timestamp=1000,
- )
- await sync_client.append_payload(
- starfish_client,
- push_path=_push_path(),
- payload=second_signal,
- timestamp=2000,
- )
- items = await starfish_client.pull(_pull_path(), append_field="items", full=True)
- payloads = [
- element["data"] if isinstance(element, dict) and "data" in element else element
- for element in items
- ]
- assert first_signal in payloads
- assert second_signal in payloads
diff --git a/packages/sync/tests/test_server.py b/packages/sync/tests/test_server.py
index 600e7d0551..ee859a4d24 100644
--- a/packages/sync/tests/test_server.py
+++ b/packages/sync/tests/test_server.py
@@ -198,26 +198,6 @@ async def test_unmatched_collection_returns_none_when_no_stored_value(self):
result = await server.get_data("users/0xwallet/settings", context)
assert result is None
- @pytest.mark.asyncio
- async def test_product_signals_returns_stored_document_unchanged(self):
- stored_document = json.dumps({
- "v": 1,
- "data": {"items": [{"ts": 1, "data": {"strategy_id": "prod-1"}}]},
- "hash": "append-hash",
- })
- mock_store = mock.MagicMock()
- mock_store.get_string = mock.AsyncMock(return_value=stored_document)
- context = _make_context(
- collection=enums.TemporaryCollections.TEMP_PRODUCT_SIGNALS.value,
- )
- with mock.patch("octobot_sync.server._get_opaque_store", return_value=mock_store):
- result = await server.get_data(
- "products/prod-1/1.0.0/signals", context
- )
- mock_store.get_string.assert_awaited_once_with("products/prod-1/1.0.0/signals")
- assert result == stored_document
- assert isinstance(json.loads(result)["data"], dict)
-
class TestUserActionsAfterWrite:
@pytest.mark.asyncio
@@ -324,25 +304,6 @@ async def test_put_data_accepts_dict_payload_from_client_encryptor(self):
assert stored_key == "users/0xwallet/accounts"
assert json.loads(stored_value) == blob
- @pytest.mark.asyncio
- async def test_product_signals_stores_full_body_verbatim(self):
- body = json.dumps({
- "v": 1,
- "data": {"items": []},
- "ts": 1000,
- "hash": "append-hash",
- })
- mock_store = mock.MagicMock()
- mock_store.put = mock.AsyncMock()
- context = _make_context(
- collection=enums.TemporaryCollections.TEMP_PRODUCT_SIGNALS.value,
- )
- with mock.patch("octobot_sync.server._get_opaque_store", return_value=mock_store):
- await server.put_data("products/prod-1/1.0.0/signals", body, context)
- mock_store.put.assert_awaited_once_with(
- "products/prod-1/1.0.0/signals", body, content_type="application/json"
- )
-
class TestStoredDocumentHelpers:
def test_wrap_produces_stored_document_shape(self):
@@ -564,3 +525,18 @@ def test_returns_app(self):
call_kwargs = mock_sync_app.create_app.call_args
assert call_kwargs.kwargs["is_allowed_user_id"] is None
assert call_kwargs.kwargs["sync_config"] is None
+ assert call_kwargs.kwargs["role_enricher"] is None
+
+ def test_forwards_role_enricher(self):
+ sentinel_app = mock.MagicMock()
+ sentinel_role_enricher = mock.MagicMock()
+ with (
+ mock.patch("octobot_sync.server.build_object_store", return_value=mock.MagicMock()) as mock_build,
+ mock.patch("octobot_sync.server.sync_app") as mock_sync_app,
+ ):
+ mock_sync_app.create_app.return_value = sentinel_app
+ result = server.build_default_sync_app(role_enricher=sentinel_role_enricher)
+ assert result is sentinel_app
+ mock_build.assert_called_once()
+ call_kwargs = mock_sync_app.create_app.call_args
+ assert call_kwargs.kwargs["role_enricher"] is sentinel_role_enricher
diff --git a/packages/sync/tests/test_sync_collections.py b/packages/sync/tests/test_sync_collections.py
index db454dd01f..937030c7d8 100644
--- a/packages/sync/tests/test_sync_collections.py
+++ b/packages/sync/tests/test_sync_collections.py
@@ -87,8 +87,7 @@ def test_fallback_to_default_config():
assert config.version == 1
assert config.namespaces is not None
ns_collections = config.namespaces["octobot"].collections
- # 8 standard user collections + 1 temporary product-scoped append-only log.
- assert len(ns_collections) == 10
+ assert len(ns_collections) == 9
by_name = {c.name: c for c in ns_collections}
assert set(by_name) == {
"user-data",
@@ -100,7 +99,6 @@ def test_fallback_to_default_config():
"user-strategies",
"user-actions",
"debug",
- "product-signals",
}
assert by_name["user-data"].storage_path == "users/{identity}/data"
assert by_name["user-accounts"].storage_path == "users/{identity}/accounts"
@@ -111,23 +109,10 @@ def test_fallback_to_default_config():
assert by_name["user-strategies"].storage_path == "users/{identity}/strategies"
assert by_name["user-actions"].storage_path == "users/{identity}/actions"
assert by_name["debug"].storage_path == "users/{identity}/debug"
- # The 8 standard user collections are "self"-scoped and "delegated"; the
- # temporary product-signals log is a product-scoped (device:root) plaintext
- # append-only by_timestamp collection.
- for name, col in by_name.items():
- if name == "product-signals":
- continue
+ for col in by_name.values():
assert col.read_roles == ["self"]
assert col.write_roles == ["self"]
assert col.encryption == "delegated"
- signals = by_name["product-signals"]
- assert signals.storage_path == "products/{product_id}/{version}/signals"
- assert signals.read_roles == ["device:root"]
- assert signals.write_roles == ["device:root"]
- assert signals.encryption == "none"
- assert signals.append_only is not None
- assert signals.append_only.type == "by_timestamp"
- assert signals.append_only.require_author_signature is False
actions = by_name["user-actions"]
assert actions.append_only is not None
assert actions.append_only.type == "by_timestamp"
diff --git a/tests/unit_tests/community/test_authentication.py b/tests/unit_tests/community/test_authentication.py
index 167acc785f..45679fb5d8 100644
--- a/tests/unit_tests/community/test_authentication.py
+++ b/tests/unit_tests/community/test_authentication.py
@@ -32,6 +32,7 @@
pytestmark = pytest.mark.asyncio
AUTH_URL = "https://oh.fake/auth"
+_TEST_SYNC_URL = "https://test-sync.example"
AUTH_RETURN = {
"access_token": "1",
"refresh_token": "2",
@@ -740,3 +741,134 @@ async def test_non_process_child_sets_pending_flag_only(self):
has_tentacles_mock.assert_awaited_once_with(auth, install_only=False)
assert auth.user_account.has_pending_packages_to_install is True
+
+def test_sync_server_url_is_a_bare_origin():
+ # octobot_sync.client/mirror.writer append SYNC_MOUNT_PATH themselves; a pre-suffixed
+ # default here doubles the "/sync" segment and 404s before reaching the sync server.
+ assert not constants.SYNC_SERVER_URL.rstrip("/").endswith("/sync")
+ assert not constants.STAGING_SYNC_SERVER_URL.rstrip("/").endswith("/sync")
+
+
+def _new_auth_for_signal_session_tests():
+ auth = community.CommunityAuthentication.__new__(community.CommunityAuthentication)
+ auth._dk_sessions = {}
+ auth._dk_session_lock = asyncio.Lock()
+ return auth
+
+
+class TestGetSessionForAddress:
+ async def test_returns_the_same_session_on_a_second_call_for_the_same_address(self):
+ auth = _new_auth_for_signal_session_tests()
+ auth.get_wallet = mock.Mock(return_value=mock.Mock(private_key="pk"))
+ sentinel_session = mock.Mock()
+ with (
+ mock.patch.object(
+ octobot.community.authentication.identifiers_provider.IdentifiersProvider,
+ "SYNC_SERVER_URL",
+ _TEST_SYNC_URL,
+ ),
+ mock.patch.object(
+ octobot.community.authentication.sync_session_writer,
+ "derived_identity_for_mirror",
+ return_value="derived-identity",
+ ),
+ mock.patch.object(
+ octobot.community.authentication.sync_session_writer,
+ "build_mirror_session",
+ mock.AsyncMock(return_value=sentinel_session),
+ ) as mock_build_mirror_session,
+ ):
+ first = await auth.get_session_for_address("0xabc")
+ second = await auth.get_session_for_address("0xabc")
+
+ assert first is sentinel_session
+ assert second is sentinel_session
+ mock_build_mirror_session.assert_awaited_once()
+
+ async def test_raises_wallet_error_when_sync_server_url_is_not_configured(self):
+ auth = _new_auth_for_signal_session_tests()
+ with mock.patch.object(
+ octobot.community.authentication.identifiers_provider.IdentifiersProvider,
+ "SYNC_SERVER_URL",
+ "",
+ ):
+ with pytest.raises(octobot.community.wallet_backend.WalletError):
+ await auth.get_session_for_address("0xabc")
+
+ async def test_builds_the_session_with_the_bare_sync_url_and_signal_name(self):
+ auth = _new_auth_for_signal_session_tests()
+ auth.get_wallet = mock.Mock(return_value=mock.Mock(private_key="pk"))
+ with (
+ mock.patch.object(
+ octobot.community.authentication.identifiers_provider.IdentifiersProvider,
+ "SYNC_SERVER_URL",
+ _TEST_SYNC_URL,
+ ),
+ mock.patch.object(
+ octobot.community.authentication.sync_session_writer,
+ "derived_identity_for_mirror",
+ return_value="derived-identity",
+ ) as mock_derive,
+ mock.patch.object(
+ octobot.community.authentication.sync_session_writer,
+ "build_mirror_session",
+ mock.AsyncMock(return_value=mock.Mock()),
+ ) as mock_build_mirror_session,
+ ):
+ await auth.get_session_for_address("0xabc")
+
+ mock_derive.assert_called_once_with("pk")
+ mock_build_mirror_session.assert_awaited_once_with(
+ "derived-identity", _TEST_SYNC_URL, name="octobot-signals"
+ )
+
+ async def test_distinct_addresses_get_distinct_sessions(self):
+ auth = _new_auth_for_signal_session_tests()
+ auth.get_wallet = mock.Mock(return_value=mock.Mock(private_key="pk"))
+ sessions = [mock.Mock(), mock.Mock()]
+ with (
+ mock.patch.object(
+ octobot.community.authentication.identifiers_provider.IdentifiersProvider,
+ "SYNC_SERVER_URL",
+ _TEST_SYNC_URL,
+ ),
+ mock.patch.object(
+ octobot.community.authentication.sync_session_writer,
+ "derived_identity_for_mirror",
+ return_value="derived-identity",
+ ),
+ mock.patch.object(
+ octobot.community.authentication.sync_session_writer,
+ "build_mirror_session",
+ mock.AsyncMock(side_effect=sessions),
+ ),
+ ):
+ first = await auth.get_session_for_address("0xabc")
+ second = await auth.get_session_for_address("0xdef")
+
+ assert first is sessions[0]
+ assert second is sessions[1]
+ assert first is not second
+
+
+class TestStopClosesCachedDkSessions:
+ async def test_stop_closes_and_clears_cached_dk_sessions(self):
+ auth = community.CommunityAuthentication.__new__(community.CommunityAuthentication)
+ auth.logger = mock.Mock()
+ auth._fetch_account_task = None
+ auth.supabase_client = mock.Mock(aclose=mock.AsyncMock())
+ auth._community_feed = None
+ auth.community_bot = None
+ auth._sync_client = None
+ cached_session = mock.Mock(
+ content_client=mock.Mock(close=mock.AsyncMock()),
+ account_client=mock.Mock(close=mock.AsyncMock()),
+ )
+ auth._dk_sessions = {"0xabc": cached_session}
+
+ await auth.stop()
+
+ cached_session.content_client.close.assert_awaited_once()
+ cached_session.account_client.close.assert_awaited_once()
+ assert auth._dk_sessions == {}
+