From 9b11d667f1fb4f4b3d4b360bdb22693ee393d165 Mon Sep 17 00:00:00 2001 From: glatinone <93207632+glatinone@users.noreply.github.com> Date: Sun, 4 Oct 2026 15:53:22 +0800 Subject: [PATCH] test: one way to install app state, and a CI run that cannot hide a leak Two test files reached the HTTP app through state a *previous* file had left on a module global. `tests/test_memory_crud.py` failed four of its own tests and `tests/test_error_shape.py` two when run on their own - and they pass in a single-process run, which is what CI does and what this session had been doing. A test that only passes in the company of another test is not measuring what it claims to. Root cause is the shape, not the two files: five files had each grown their own slightly different installer (`_install_app`, `_client`, `_install_app_state`, an autouse fixture), because nothing shared existed. That is now one function. - `conftest.install_app_state(**overrides)` installs fresh storage, engine, key store and scoring limiter, and returns what it installed; `app_state` and `app_client` fixtures build on it. It takes the knobs the files actually vary - `api_key_store`, `scoring_limit`, `scoring_clock`, `lifecycle_settings`, `retention_days` - so `test_auth`, `test_ratelimit`, `test_scheduler` keep their specifics in a two-line adapter instead of a ten-line copy. - Six files migrated; the duplicated installers are gone. `test_transitions`' three HTTP tests also lost the `created_at` they echoed back, which the previous PR made unnecessary. - **CI runs the server suite one process per file.** The shell already runs with `-e`, so the first failing file stops the build. This is the discipline `scripts/run_tests.sh` enforces in the project this repo borrows its test practice from, and it is what makes the fix stick: without it, someone reintroduces a cross-file dependency and the build stays green because the order happens to work. - `CONTRIBUTING.md` states the rule and shows the fixture pattern, so the next file has an obvious right way to do it rather than a blank page. Verified with the CI loop itself, run locally: all 16 files pass alone, 244 tests, 26 skipped (the Postgres half runs in its own job). --- .github/CONTRIBUTING.md | 26 +++++ .github/workflows/ci.yml | 15 ++- CHANGELOG.md | 9 ++ server/tests/conftest.py | 70 +++++++++++++ server/tests/test_auth.py | 13 +-- server/tests/test_error_shape.py | 46 ++++----- server/tests/test_memory_crud.py | 19 +--- server/tests/test_paging.py | 31 +----- server/tests/test_ratelimit.py | 22 +--- server/tests/test_scheduler.py | 15 ++- server/tests/test_spec_capabilities.py | 14 +-- server/tests/test_transitions.py | 133 ++++++++----------------- 12 files changed, 210 insertions(+), 203 deletions(-) diff --git a/.github/CONTRIBUTING.md b/.github/CONTRIBUTING.md index 56e0233..fec43ca 100644 --- a/.github/CONTRIBUTING.md +++ b/.github/CONTRIBUTING.md @@ -71,6 +71,32 @@ pytest -v python/tests node --test test/client.test.js ``` +### Tests + +**Every test file must pass on its own.** CI runs one process per file, so a test +that only passes alongside another file fails the build rather than hiding. That +rule exists because it has been broken twice: `test_memory_crud.py` and +`test_error_shape.py` both reached the HTTP app through state a previous file had +left on a module global, which held in a single-process run and broke the moment +one file ran alone. + +A file that talks to the HTTP app takes the fixtures from `conftest.py`: + +```python +@pytest.fixture(autouse=True) +def _state(): + install_app_state() # fresh storage, engine, key store, limiter + + +async def test_something(app_client): # AsyncClient on that state + ... +``` + +`install_app_state(...)` takes the knobs a file needs to vary - `api_key_store`, +`scoring_limit`, `scoring_clock`, `lifecycle_settings`, `retention_days` - and +returns what it installed. Never install state from a test body that the next test +depends on. + ### Storage backends `tests/test_adapter_contract.py` runs the same behaviour tests against every diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7f4a238..3f9d0fb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,9 +52,20 @@ jobs: - name: Install server package working-directory: server run: pip install -e ".[dev]" - - name: Run server tests + # One process per test file - the discipline `scripts/run_tests.sh` enforces + # in the project this repo borrows its test practice from, and for the same + # reason. Two files here reached the app through state a *previous* file had + # left on the module: they passed in a single-process run and failed the + # moment one file was run alone. The shell already runs with `-e`, so the + # loop stops at the first failing file. + - name: Run server tests (one process per file) working-directory: server - run: pytest -v + run: | + for f in tests/test_*.py; do + echo "::group::$f" + pytest -q "$f" + echo "::endgroup::" + done - name: Security scan (pip-audit) working-directory: server continue-on-error: true diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d10239..1d43c8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -218,6 +218,15 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). module, so the file only passed as part of the whole suite. An autouse fixture gives each test its own state. +- **One place installs app state for tests, and CI runs one process per test + file.** Two files (`test_memory_crud.py`, `test_error_shape.py`) reached the HTTP + app through state a *previous* file had left on a module global. They passed as + part of the whole suite and failed the moment a single file was run - which is + how anyone runs one while working on it. Five files had each grown their own + slightly different installer; that is now `conftest.install_app_state()` with + `app_state` / `app_client` fixtures, and CI runs the server suite file by file so + the whole class fails loudly instead of being averaged away by test ordering. + ### Changed - **Every endpoint returns one error shape.** `PATCH /memories/{id}` answered a conflict with `{"detail": ...}` while `DELETE` answered with diff --git a/server/tests/conftest.py b/server/tests/conftest.py index 5c0fe52..bd7c2f4 100644 --- a/server/tests/conftest.py +++ b/server/tests/conftest.py @@ -5,6 +5,7 @@ import os import uuid from datetime import UTC, datetime +from types import SimpleNamespace import pytest @@ -22,6 +23,7 @@ OwnerType, SourceType, ) +from amp_server.retention import RETENTION_DAYS from amp_server.storage.chroma import ChromaAdapter @@ -119,6 +121,74 @@ def make_cell( ) +# --- The HTTP app ----------------------------------------------------------- +# +# Every test file that talks to the HTTP app needs the app pointed at state the +# test owns, because `get_storage()` reads a module global that the lifespan sets +# when a real server boots. Each file used to install that itself - five slightly +# different copies - and a file that forgot reached whatever state a *previous +# file* had left behind. That held while the whole suite ran in one process and +# broke the moment a single file was run, which is how anyone runs it while +# working. One helper, in conftest, so a file cannot get it subtly differently and +# cannot silently inherit somebody else's. + + +def install_app_state( + *, + api_key_store=None, + scoring_limit=None, + scoring_clock=None, + lifecycle_settings=None, + retention_days: int = RETENTION_DAYS, +): + """Point the app at fresh state the way the lifespan would. + + Returns the objects it installed, so a test can assert against the same + storage or limiter the app is using rather than a second copy. + """ + import amp_server.main as main_mod + from amp_server.lifecycle import LifecycleEngine + from amp_server.ratelimit import ScoringPatchLimit, ScoringPatchLimiter + + storage = ChromaAdapter( + collection_name=f"test_{uuid.uuid4().hex[:12]}", + retention_days=retention_days, + ) + limit = scoring_limit or ScoringPatchLimit() + limiter = ScoringPatchLimiter( + limit, **({"clock": scoring_clock} if scoring_clock else {}) + ) + + main_mod._storage = storage + main_mod._lifecycle = LifecycleEngine(storage) + main_mod._api_key_store = api_key_store + main_mod._scoring_limit = limit + main_mod._scoring_limiter = limiter + if lifecycle_settings is not None: + main_mod._lifecycle_settings = lifecycle_settings + + return SimpleNamespace(storage=storage, limiter=limiter, scoring_limit=limit) + + +@pytest.fixture +def app_state(): + """Fresh app state for one test.""" + return install_app_state() + + +@pytest.fixture +async def app_client(app_state): + """An AsyncClient over the app, on state this test owns.""" + from httpx import ASGITransport, AsyncClient + + import amp_server.main as main_mod + + async with AsyncClient( + transport=ASGITransport(app=main_mod.app), base_url="http://test" + ) as client: + yield client + + def make_create_body( *, owner_id: str = "user-123", diff --git a/server/tests/test_auth.py b/server/tests/test_auth.py index d98f296..03de6d3 100644 --- a/server/tests/test_auth.py +++ b/server/tests/test_auth.py @@ -14,10 +14,9 @@ from __future__ import annotations import json -import uuid import pytest -from conftest import make_cell +from conftest import install_app_state, make_cell from httpx import ASGITransport, AsyncClient from amp_server.auth import ApiKeyStore, digest, load_store, store_from_env @@ -28,14 +27,8 @@ def _install_app(store: ApiKeyStore | None = None) -> None: - """Point the app at a fresh storage + engine, as the lifespan would.""" - import amp_server.main as main_mod - from amp_server.lifecycle import LifecycleEngine - from amp_server.storage.chroma import ChromaAdapter - - main_mod._storage = ChromaAdapter(collection_name=f"test_{uuid.uuid4().hex[:12]}") - main_mod._lifecycle = LifecycleEngine(main_mod._storage) - main_mod._api_key_store = store + """Fresh state with this file's one difference: the key store.""" + install_app_state(api_key_store=store) def _store(monkeypatch, tmp_path, keys: dict[str, str]) -> ApiKeyStore: diff --git a/server/tests/test_error_shape.py b/server/tests/test_error_shape.py index 90394ae..0056afe 100644 --- a/server/tests/test_error_shape.py +++ b/server/tests/test_error_shape.py @@ -6,12 +6,16 @@ to a generic "HTTP error 409" in a client. Every protocol error now comes from `amp_server.errors.AMPError`, and these tests fix the wire contract so a future inline `JSONResponse` cannot quietly introduce a second shape. + +These tests used to build their own client against the bare app, which meant they +reached whatever storage a *previous test file* had left on the module - so half of +them failed when this file ran on its own. They take `app_client` from conftest now, +which installs state this test owns. """ from __future__ import annotations import pytest -from httpx import ASGITransport, AsyncClient from amp_server.errors import ( AMPError, @@ -41,45 +45,31 @@ def _assert_error_envelope(body: dict, expected_code: str) -> None: @pytest.mark.asyncio -async def test_missing_agent_id_uses_the_error_envelope(): - from amp_server.main import app - - async with AsyncClient( - transport=ASGITransport(app=app), base_url="http://test" - ) as client: - resp = await client.post("/amp/v1/memories", json=_minimal_body()) +async def test_missing_agent_id_uses_the_error_envelope(app_client): + resp = await app_client.post("/amp/v1/memories", json=_minimal_body()) assert resp.status_code == 401 _assert_error_envelope(resp.json(), "MISSING_AGENT_ID") @pytest.mark.asyncio -async def test_access_denied_uses_the_error_envelope(): - from amp_server.main import app - - async with AsyncClient( - transport=ASGITransport(app=app), base_url="http://test" - ) as client: - resp = await client.get("/amp/v1/memories/does-not-exist", headers=_HEADERS) +async def test_access_denied_uses_the_error_envelope(app_client): + resp = await app_client.get("/amp/v1/memories/does-not-exist", headers=_HEADERS) assert resp.status_code == 403 _assert_error_envelope(resp.json(), "ACCESS_DENIED") @pytest.mark.asyncio -async def test_invalid_transition_uses_the_error_envelope(): - from amp_server.main import app - - async with AsyncClient( - transport=ASGITransport(app=app), base_url="http://test" - ) as client: - created = await client.post( - "/amp/v1/memories", headers=_HEADERS, json=_minimal_body() - ) - assert created.status_code == 201 - resp = await client.delete( - f"/amp/v1/memories/{created.json()['id']}", headers=_HEADERS - ) +async def test_invalid_transition_uses_the_error_envelope(app_client): + created = await app_client.post( + "/amp/v1/memories", headers=_HEADERS, json=_minimal_body() + ) + assert created.status_code == 201 + + resp = await app_client.delete( + f"/amp/v1/memories/{created.json()['id']}", headers=_HEADERS + ) assert resp.status_code == 409 _assert_error_envelope(resp.json(), "INVALID_TRANSITION") diff --git a/server/tests/test_memory_crud.py b/server/tests/test_memory_crud.py index 0c6ae3e..ae4894e 100644 --- a/server/tests/test_memory_crud.py +++ b/server/tests/test_memory_crud.py @@ -6,7 +6,7 @@ from datetime import UTC, datetime, timedelta import pytest -from conftest import make_cell +from conftest import install_app_state, make_cell from httpx import ASGITransport, AsyncClient from amp_server.models import ( @@ -28,20 +28,11 @@ def _app_state(): The HTTP tests here used to reach the app through whatever storage a *previous test file* had left on the module. That held while the whole suite ran in one - process and broke the moment this file ran alone - which is how CI runs it in - the Hermes repository this project borrows its test discipline from, and how - anyone runs a single file while working on it. A test that only passes in the - company of another test is not measuring what it claims to. + process and broke the moment this file ran alone - which is how a single file + gets run while working on it. A test that only passes in the company of another + test is not measuring what it claims to. """ - import amp_server.main as main_mod - from amp_server.lifecycle import LifecycleEngine - from amp_server.ratelimit import ScoringPatchLimit, ScoringPatchLimiter - - main_mod._storage = ChromaAdapter(collection_name=f"test_{uuid.uuid4().hex[:12]}") - main_mod._lifecycle = LifecycleEngine(main_mod._storage) - main_mod._api_key_store = None - main_mod._scoring_limit = ScoringPatchLimit() - main_mod._scoring_limiter = ScoringPatchLimiter(main_mod._scoring_limit) + install_app_state() # --------------------------------------------------------------------------- diff --git a/server/tests/test_paging.py b/server/tests/test_paging.py index b99415f..81f299b 100644 --- a/server/tests/test_paging.py +++ b/server/tests/test_paging.py @@ -10,10 +10,8 @@ from __future__ import annotations -import uuid - import pytest -from conftest import make_cell +from conftest import install_app_state, make_cell from httpx import ASGITransport, AsyncClient, Response from amp_server.models import MAX_PAGE_SIZE, LifecycleStatus @@ -23,14 +21,10 @@ _STRANGER = "agent-paging-stranger" -def _install_app() -> None: - import amp_server.main as main_mod - from amp_server.lifecycle import LifecycleEngine - from amp_server.storage.chroma import ChromaAdapter - - main_mod._storage = ChromaAdapter(collection_name=f"test_{uuid.uuid4().hex[:12]}") - main_mod._lifecycle = LifecycleEngine(main_mod._storage) - main_mod._api_key_store = None +@pytest.fixture(autouse=True) +def _state(): + """Every test here talks to the app, so every test gets its own state.""" + install_app_state() async def _client() -> AsyncClient: @@ -64,7 +58,6 @@ def _headers(agent_id: str = _READER) -> dict[str, str]: @pytest.mark.asyncio async def test_a_page_reports_what_it_returned_and_whether_more_follows(): - _install_app() await _seed(3, readable_by=[_READER]) async with await _client() as client: @@ -85,7 +78,6 @@ async def test_a_page_reports_what_it_returned_and_whether_more_follows(): @pytest.mark.asyncio async def test_pages_do_not_overlap_and_cover_everything(): - _install_app() stored = await _seed(5, readable_by=[_READER]) seen: list[str] = [] @@ -114,7 +106,6 @@ async def test_the_limit_counts_cells_the_caller_may_read(): must be three readable cells with `has_more` false - the old store-level limit would have examined only the first three candidates and returned none. """ - _install_app() await _seed(4, readable_by=[_STRANGER]) readable = await _seed(3, readable_by=[_READER]) @@ -128,7 +119,6 @@ async def test_the_limit_counts_cells_the_caller_may_read(): @pytest.mark.asyncio async def test_offset_past_the_end_is_empty_and_says_so(): - _install_app() await _seed(2, readable_by=[_READER]) async with await _client() as client: @@ -143,7 +133,6 @@ async def test_offset_past_the_end_is_empty_and_says_so(): @pytest.mark.asyncio async def test_a_cell_the_caller_cannot_read_never_shows_up(): - _install_app() await _seed(3, readable_by=[_STRANGER]) async with await _client() as client: @@ -154,7 +143,6 @@ async def test_a_cell_the_caller_cannot_read_never_shows_up(): @pytest.mark.asyncio async def test_deleted_cells_are_not_listed_by_default(): - _install_app() import amp_server.main as main_mod cell = make_cell( @@ -178,7 +166,6 @@ async def test_deleted_cells_are_not_listed_by_default(): @pytest.mark.asyncio async def test_the_page_size_is_bounded_on_both_ends(): """An unbounded limit is a request for the whole store.""" - _install_app() await _seed(1, readable_by=[_READER]) async with await _client() as client: @@ -197,7 +184,6 @@ async def test_the_page_size_is_bounded_on_both_ends(): @pytest.mark.asyncio async def test_the_query_alias_pages_the_same_way(): - _install_app() await _seed(3, readable_by=[_READER]) async with await _client() as client: @@ -214,7 +200,6 @@ async def test_the_query_alias_pages_the_same_way(): @pytest.mark.asyncio async def test_spec_advertises_the_page_ceiling(): - _install_app() async with await _client() as client: capabilities = (await client.get("/amp/v1/spec")).json()["capabilities"] @@ -243,7 +228,6 @@ def _query(**overrides: object) -> dict: @pytest.mark.asyncio async def test_search_pages_with_offset_and_reports_has_more(): - _install_app() await _seed(3, readable_by=[_READER]) first = (await _search(_query(limit=2))).json() @@ -257,7 +241,6 @@ async def test_search_pages_with_offset_and_reports_has_more(): @pytest.mark.asyncio async def test_search_echoes_the_window_it_used(): - _install_app() await _seed(2, readable_by=[_READER]) body = (await _search(_query(limit=1, offset=1))).json() @@ -269,7 +252,6 @@ async def test_search_echoes_the_window_it_used(): @pytest.mark.asyncio async def test_search_pages_do_not_overlap_and_cover_everything(): - _install_app() stored = await _seed(5, readable_by=[_READER]) seen: list[str] = [] @@ -295,7 +277,6 @@ async def test_search_pages_do_not_overlap_and_cover_everything(): @pytest.mark.asyncio async def test_the_search_window_counts_results_the_caller_may_read(): """Same rule as the listing endpoints: read first, then window.""" - _install_app() await _seed(4, readable_by=[_STRANGER]) readable = await _seed(3, readable_by=[_READER]) @@ -308,7 +289,6 @@ async def test_the_search_window_counts_results_the_caller_may_read(): @pytest.mark.asyncio async def test_a_search_window_outside_the_bounds_is_refused(): - _install_app() await _seed(1, readable_by=[_READER]) assert (await _search(_query(limit=MAX_PAGE_SIZE + 1))).status_code == 422 @@ -319,7 +299,6 @@ async def test_a_search_window_outside_the_bounds_is_refused(): @pytest.mark.asyncio async def test_the_search_ceiling_is_the_advertised_one(): """One number for both endpoints, so a client cannot be told two.""" - _install_app() async with await _client() as client: advertised = (await client.get("/amp/v1/spec")).json()["capabilities"][ diff --git a/server/tests/test_ratelimit.py b/server/tests/test_ratelimit.py index dee0e9d..95db5a2 100644 --- a/server/tests/test_ratelimit.py +++ b/server/tests/test_ratelimit.py @@ -10,10 +10,8 @@ from __future__ import annotations -import uuid - import pytest -from conftest import make_cell +from conftest import install_app_state, make_cell from httpx import ASGITransport, AsyncClient, Response from amp_server.models import LifecycleStatus, MemoryCellUpdate, MemoryScoring @@ -45,20 +43,10 @@ def advance(self, seconds: float) -> None: def _install_app( limit: ScoringPatchLimit | None = None, clock: FakeClock | None = None ) -> ScoringPatchLimiter: - """Fresh storage plus a limiter the test controls.""" - import amp_server.main as main_mod - from amp_server.lifecycle import LifecycleEngine - from amp_server.storage.chroma import ChromaAdapter - - main_mod._storage = ChromaAdapter(collection_name=f"test_{uuid.uuid4().hex[:12]}") - main_mod._lifecycle = LifecycleEngine(main_mod._storage) - main_mod._api_key_store = None - - resolved = limit or ScoringPatchLimit() - limiter = ScoringPatchLimiter(resolved, clock=clock or FakeClock()) - main_mod._scoring_limit = resolved - main_mod._scoring_limiter = limiter - return limiter + """Fresh state with the budget and clock this test wants.""" + return install_app_state( + scoring_limit=limit, scoring_clock=clock or FakeClock() + ).limiter async def _client() -> AsyncClient: diff --git a/server/tests/test_scheduler.py b/server/tests/test_scheduler.py index 28bd46d..430a215 100644 --- a/server/tests/test_scheduler.py +++ b/server/tests/test_scheduler.py @@ -7,7 +7,7 @@ from datetime import UTC, datetime, timedelta import pytest -from conftest import make_cell +from conftest import install_app_state, make_cell from httpx import ASGITransport, AsyncClient from amp_server.lifecycle import LifecycleEngine @@ -315,13 +315,12 @@ def _install_app_state( retention_days: int = RETENTION_DAYS, purge_retention: bool = False, ) -> None: - """Point the app at a fresh storage + engine, as the lifespan would.""" - import amp_server.main as main_mod - - main_mod._storage = _fresh_storage(retention_days=retention_days) - main_mod._lifecycle = LifecycleEngine(main_mod._storage) - main_mod._lifecycle_settings = _settings( - admin_token=token, purge_retention=purge_retention + """Fresh state with this file's difference: the lifecycle settings.""" + install_app_state( + retention_days=retention_days, + lifecycle_settings=_settings( + admin_token=token, purge_retention=purge_retention + ), ) diff --git a/server/tests/test_spec_capabilities.py b/server/tests/test_spec_capabilities.py index 151a94b..0c3340e 100644 --- a/server/tests/test_spec_capabilities.py +++ b/server/tests/test_spec_capabilities.py @@ -8,10 +8,8 @@ from __future__ import annotations -import uuid - import pytest -from conftest import make_cell +from conftest import install_app_state, make_cell from httpx import ASGITransport, AsyncClient from amp_server.errors import AMPError @@ -28,13 +26,15 @@ def _body(text: str) -> dict: } +@pytest.fixture(autouse=True) +def _state(): + """Every test here reads `/spec`, which reports the state the app is holding.""" + install_app_state() + + async def _client(): import amp_server.main as main_mod - from amp_server.lifecycle import LifecycleEngine - from amp_server.storage.chroma import ChromaAdapter - main_mod._storage = ChromaAdapter(collection_name=f"test_{uuid.uuid4().hex[:12]}") - main_mod._lifecycle = LifecycleEngine(main_mod._storage) return AsyncClient( transport=ASGITransport(app=main_mod.app), base_url="http://test" ) diff --git a/server/tests/test_transitions.py b/server/tests/test_transitions.py index 0fe1efb..2e2abe8 100644 --- a/server/tests/test_transitions.py +++ b/server/tests/test_transitions.py @@ -13,7 +13,6 @@ import pytest from conftest import make_cell -from httpx import ASGITransport, AsyncClient from amp_server.lifecycle import check_status_transition from amp_server.models import ( @@ -149,109 +148,61 @@ async def _create(client) -> dict: @pytest.mark.asyncio -async def test_patch_to_deleted_is_409(): - import amp_server.main as main_mod - from amp_server.lifecycle import LifecycleEngine - from amp_server.storage.chroma import ChromaAdapter - - main_mod._storage = ChromaAdapter(collection_name=f"test_{uuid.uuid4().hex[:12]}") - main_mod._lifecycle = LifecycleEngine(main_mod._storage) - - async with AsyncClient( - transport=ASGITransport(app=main_mod.app), base_url="http://test" - ) as client: - cell = await _create(client) - response = await client.patch( - f"/amp/v1/memories/{cell['id']}", - headers=_HEADERS, - json={ - "lifecycle": { - "created_at": cell["lifecycle"]["created_at"], - "status": "deleted", - } - }, - ) +async def test_patch_to_deleted_is_409(app_client): + cell = await _create(app_client) + response = await app_client.patch( + f"/amp/v1/memories/{cell['id']}", + headers=_HEADERS, + json={"lifecycle": {"status": "deleted"}}, + ) assert response.status_code == 409 assert response.json()["error"]["code"] == "INVALID_TRANSITION" @pytest.mark.asyncio -async def test_archived_cell_cannot_be_patched_back_to_active(): - import amp_server.main as main_mod - from amp_server.lifecycle import LifecycleEngine - from amp_server.storage.chroma import ChromaAdapter - - main_mod._storage = ChromaAdapter(collection_name=f"test_{uuid.uuid4().hex[:12]}") - main_mod._lifecycle = LifecycleEngine(main_mod._storage) - - async with AsyncClient( - transport=ASGITransport(app=main_mod.app), base_url="http://test" - ) as client: - cell = await _create(client) - - archived = await client.patch( - f"/amp/v1/memories/{cell['id']}", - headers=_HEADERS, - json={ - "lifecycle": { - "created_at": cell["lifecycle"]["created_at"], - "status": "archived", - } - }, - ) - assert archived.status_code == 200 - - resurrect = await client.patch( - f"/amp/v1/memories/{cell['id']}", - headers=_HEADERS, - json={ - "lifecycle": { - "created_at": archived.json()["lifecycle"]["created_at"], - "status": "active", - } - }, - ) +async def test_archived_cell_cannot_be_patched_back_to_active(app_client): + cell = await _create(app_client) + + archived = await app_client.patch( + f"/amp/v1/memories/{cell['id']}", + headers=_HEADERS, + json={"lifecycle": {"status": "archived"}}, + ) + assert archived.status_code == 200 + + resurrect = await app_client.patch( + f"/amp/v1/memories/{cell['id']}", + headers=_HEADERS, + json={"lifecycle": {"status": "active"}}, + ) assert resurrect.status_code == 409 assert resurrect.json()["error"]["code"] == "INVALID_TRANSITION" @pytest.mark.asyncio -async def test_a_deleted_cell_is_403_not_409(): +async def test_a_deleted_cell_is_403_not_409(app_client): """ยง8.4 still wins: a deleted cell is invisible, not explained.""" - import amp_server.main as main_mod - from amp_server.lifecycle import LifecycleEngine - from amp_server.storage.chroma import ChromaAdapter - - main_mod._storage = ChromaAdapter(collection_name=f"test_{uuid.uuid4().hex[:12]}") - main_mod._lifecycle = LifecycleEngine(main_mod._storage) - - async with AsyncClient( - transport=ASGITransport(app=main_mod.app), base_url="http://test" - ) as client: - cell = await _create(client) - # Archive then delete through the documented route. - await client.patch( - f"/amp/v1/memories/{cell['id']}", - headers=_HEADERS, - json={ - "lifecycle": { - "created_at": cell["lifecycle"]["created_at"], - "status": "archived", - } - }, - ) - deleted = await client.delete( - f"/amp/v1/memories/{cell['id']}", headers=_HEADERS - ) - assert deleted.status_code == 204 - - response = await client.patch( - f"/amp/v1/memories/{cell['id']}", - headers=_HEADERS, - json={"content": {"text": "still here?"}}, - ) + cell = await _create(app_client) + + # Archive then delete through the documented route, one request each: no + # reading the cell first to echo `created_at` back. + await app_client.patch( + f"/amp/v1/memories/{cell['id']}", + headers=_HEADERS, + json={"lifecycle": {"status": "archived"}}, + ) + deleted = await app_client.delete( + f"/amp/v1/memories/{cell['id']}", headers=_HEADERS + ) + assert deleted.status_code == 204 + + response = await app_client.patch( + f"/amp/v1/memories/{cell['id']}", + headers=_HEADERS, + json={"content": {"text": "still here?"}}, + ) assert response.status_code == 403 assert response.json()["error"]["code"] == "ACCESS_DENIED"