Merge pull request 'feat: declare every requested app to Thoth' (#8) from marioglealp/declare-apps into main

Reviewed-on: #8

src/shikigami/activities.py

@@ -12,7 +12,7 @@ from shikigami import build_creds as creds
 from shikigami.config import Settings
 from shikigami.models import AppAction, ExistingApp, RequestsSnapshot
 from shikigami.openbao import KV
-from shikigami.sources import read_existing_apps, read_requests
+from shikigami.sources import declare_apps, read_existing_apps, read_requests
 from shikigami.tokens import TokenSource
 
 
@@ -31,6 +31,17 @@ class Activities:
     async def read_existing_apps(self) -> dict[str, ExistingApp]:
         return await read_existing_apps(self._http, self._settings.thoth_url, await self._thoth_tokens.get())
 
+    @activity.defn
+    async def declare_apps(self, snapshot: RequestsSnapshot) -> dict:
+        try:
+            return await declare_apps(self._http, self._settings.thoth_url, await self._thoth_tokens.get(), snapshot)
+        except httpx2.HTTPStatusError as e:
+            if e.response.status_code in (403, 422):  # no permission, or Thoth refused the snapshot: retries won't help
+                raise ApplicationError(
+                    f"Thoth said {e.response.status_code}: {e.response.text[:500]}", non_retryable=True
+                ) from e
+            raise
+
     @activity.defn
     async def dry_run(self, action: AppAction) -> None:
         """Step 3 stops here: say what would happen. Later steps replace this with the real work, step by step."""

src/shikigami/main.py

@@ -32,7 +32,12 @@ async def run() -> None:
             client,
             task_queue=settings.task_queue,
             workflows=[RequestsSync, AppWorkflow],
-            activities=[activities.read_requests, activities.read_existing_apps, activities.dry_run],
+            activities=[
+                activities.read_requests,
+                activities.read_existing_apps,
+                activities.declare_apps,
+                activities.dry_run,
+            ],
             max_concurrent_activities=settings.max_concurrent_activities,
         )
         server = uvicorn.Server(uvicorn.Config(make_app(settings, client), host="0.0.0.0", port=settings.http_port))

src/shikigami/sources.py

@@ -37,3 +37,19 @@ async def read_existing_apps(http: httpx2.AsyncClient, thoth_url: str, token: st
         item["key"]: ExistingApp(key=item["key"], status=item["status"], template=item["attributes"].get("template"))
         for item in response.json()
     }
+
+
+async def declare_apps(http: httpx2.AsyncClient, thoth_url: str, token: str, snapshot: RequestsSnapshot) -> dict:
+    """Every requested app, as one Thoth snapshot (D52): Shikigami declares, Thoth keeps them `planned` until a scanner
+    sees them, and only Anubis retires (D53). The scan id is the requests commit, so a resend changes nothing."""
+    items = [
+        {"type": "app", "key": name, "name": name, "attributes": {"template": app.spec.template}}
+        for name, app in sorted(snapshot.apps.items())
+    ]
+    response = await http.put(
+        f"{thoth_url}/v1/scans/requests-{snapshot.commit}/shikigami/all/items",
+        json={"items": items},
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    response.raise_for_status()
+    return response.json()

src/shikigami/workflows.py

@@ -78,7 +78,13 @@ class RequestsSync:
                     Activities.read_existing_apps, start_to_close_timeout=TIMEOUT, retry_policy=RETRY
                 )
                 actions = plan(snapshot.apps, existing)
-                workflow.logger.info("requests at %s: %d app step(s)", snapshot.commit, len(actions))
+                # Planned against what Thoth had; then declare the whole request picture before the apps' steps run.
+                summary = await workflow.execute_activity_method(
+                    Activities.declare_apps, snapshot, start_to_close_timeout=TIMEOUT, retry_policy=RETRY
+                )
+                workflow.logger.info(
+                    "requests at %s: %d app step(s); Thoth: %s", snapshot.commit, len(actions), summary
+                )
                 await asyncio.gather(*(self._start_app(a) for a in actions))
             if not self._poked:
                 return actions

tests/test_activities.py

@@ -0,0 +1,37 @@
+import httpx2
+import pytest
+from pydantic import SecretStr
+from temporalio.exceptions import ApplicationError
+from temporalio.testing import ActivityEnvironment
+
+from shikigami.activities import Activities
+from shikigami.config import Settings
+from shikigami.models import RequestsSnapshot
+from shikigami.tokens import TokenSource
+
+
+def activities(status: int) -> Activities:
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if request.url.host == "auth.test":
+            return httpx2.Response(200, json={"access_token": "t", "expires_in": 600})
+        return httpx2.Response(status, json={"title": "Snapshot refused"})
+
+    http = httpx2.AsyncClient(transport=httpx2.MockTransport(handler))
+    tokens = TokenSource(http, "https://auth.test/token/", "thoth", "shikigami", SecretStr("pw"))
+    return Activities(Settings.model_construct(thoth_url="http://thoth.test"), http, tokens)
+
+
+SNAPSHOT = RequestsSnapshot(commit="c1", apps={}, databases={}, problems=[])
+
+
+@pytest.mark.parametrize("status", [403, 422])
+async def test_a_refused_snapshot_is_not_retried(status: int):
+    with pytest.raises(ApplicationError) as raised:
+        await ActivityEnvironment().run(activities(status).declare_apps, SNAPSHOT)
+    assert raised.value.non_retryable
+    assert str(status) in str(raised.value)
+
+
+async def test_a_server_error_is_retried():
+    with pytest.raises(httpx2.HTTPStatusError):
+        await ActivityEnvironment().run(activities(503).declare_apps, SNAPSHOT)

tests/test_sources.py

@@ -1,9 +1,13 @@
 import io
+import json
 import tarfile
 
 import httpx2
+import yaml
 
-from shikigami.sources import read_existing_apps, read_requests
+from shikigami.models import RequestsSnapshot
+from shikigami.request_format import App
+from shikigami.sources import declare_apps, read_existing_apps, read_requests
 
 ECHO = b"""\
 apiVersion: meru.clockmost.com/v1
@@ -67,3 +71,24 @@ async def test_thoth_apps_by_key():
     async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http:
         apps = await read_existing_apps(http, "http://thoth.test", "t")
     assert apps["echo"].template == "nuwa/python"
+
+
+async def test_declares_every_requested_app_as_one_snapshot():
+    app = App.model_validate(yaml.safe_load(ECHO))
+    seen: list[httpx2.Request] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        seen.append(request)
+        return httpx2.Response(200, json={"created": 1})
+
+    snapshot = RequestsSnapshot(commit="abc123", apps={"echo": app}, databases={}, problems=[])
+    async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http:
+        summary = await declare_apps(http, "http://thoth.test", "t", snapshot)
+    assert summary == {"created": 1}
+    [request] = seen
+    assert request.method == "PUT"
+    assert request.url.path == "/v1/scans/requests-abc123/shikigami/all/items"
+    assert request.headers["authorization"] == "Bearer t"
+    assert json.loads(request.content) == {
+        "items": [{"type": "app", "key": "echo", "name": "echo", "attributes": {"template": "nuwa/python"}}]
+    }

tests/test_workflows.py

@@ -21,20 +21,31 @@ ECHO = App.model_validate(
 )
 
 
-def activities(snapshot: RequestsSnapshot, existing: dict[str, ExistingApp], done: list[AppAction]):
+def activities(
+    snapshot: RequestsSnapshot, existing: dict[str, ExistingApp], done: list[AppAction], log: list[str] | None = None
+):
+    log = [] if log is None else log
+
     @activity.defn(name="read_requests")
     async def read_requests() -> RequestsSnapshot:
+        log.append("read_requests")
         return snapshot
 
     @activity.defn(name="read_existing_apps")
     async def read_existing_apps() -> dict[str, ExistingApp]:
+        log.append("read_existing_apps")
         return existing
 
+    @activity.defn(name="declare_apps")
+    async def declare_apps(declared: RequestsSnapshot) -> dict:
+        log.append(f"declare_apps {sorted(declared.apps)}")
+        return {"created": len(declared.apps)}
+
     @activity.defn(name="dry_run")
     async def dry_run(action: AppAction) -> None:
         done.append(action)
 
-    return [read_requests, read_existing_apps, dry_run]
+    return [read_requests, read_existing_apps, declare_apps, dry_run]
 
 
 @pytest.fixture
@@ -43,11 +54,16 @@ async def env():
         yield env
 
 
-async def run_sync(client: Client, snapshot: RequestsSnapshot, existing: dict[str, ExistingApp]):
+async def run_sync(
+    client: Client, snapshot: RequestsSnapshot, existing: dict[str, ExistingApp], log: list[str] | None = None
+):
     done: list[AppAction] = []
     queue = f"test-{uuid.uuid4()}"
     async with Worker(
-        client, task_queue=queue, workflows=[RequestsSync, AppWorkflow], activities=activities(snapshot, existing, done)
+        client,
+        task_queue=queue,
+        workflows=[RequestsSync, AppWorkflow],
+        activities=activities(snapshot, existing, done, log),
     ):
         result = await client.execute_workflow(RequestsSync.run, id=f"sync-{uuid.uuid4()}", task_queue=queue)
     return result, done
@@ -66,3 +82,18 @@ async def test_requests_with_problems_change_nothing(env: WorkflowEnvironment):
     result, done = await run_sync(env.client, snapshot, {})
     assert result == []
     assert done == []
+
+
+async def test_declares_after_planning_against_what_thoth_had(env: WorkflowEnvironment):
+    snapshot = RequestsSnapshot(commit="c1", apps={"echo": ECHO}, databases={}, problems=[])
+    log: list[str] = []
+    result, _ = await run_sync(env.client, snapshot, {}, log)
+    assert log == ["read_requests", "read_existing_apps", "declare_apps ['echo']"]
+    assert [(a.app, a.kind) for a in result] == [("echo", "register")]
+
+
+async def test_requests_with_problems_declare_nothing(env: WorkflowEnvironment):
+    snapshot = RequestsSnapshot(commit="c1", apps={"echo": ECHO}, databases={}, problems=["apps/x.yaml: broken"])
+    log: list[str] = []
+    await run_sync(env.client, snapshot, {}, log)
+    assert log == ["read_requests"]