The catalogue: declare files that already exist, without moving the bytes
The audio node is the master for the audiobooks. UmbraTrans holds the metadata needed to reason about a file and a reference to where it is; it does not hold the bytes. DECISION-6 (object storage) stays unresolved, and this is what lets the system be useful before it is resolved rather than after. POST /api/compute/artifacts takes a manifest against a project and declares each entry: content hash, size, media type, kind, and a storage_uri naming where the file actually is. GET /api/compute/artifacts lists the catalogue. There were no artifact routes at all before this - the tables existed since migration 0003 and nothing had ever written to them, so "the controller has the artifacts to show a job is complete" had nothing behind it. Every declared row lands integrity='unverified', and that is the point rather than a caveat. Ingesting is a claim: a node reporting the hash of a file is a statement about bytes nothing here has read, because there is no object storage to read them from. A design where submitting a manifest marks the catalogue verified would make the first ingest the most reassuring and least true event in the system's life. verified_at is set by reading the bytes back and hashing them, and nothing in this path does that. storage_node_id is part of the reference rather than an afterthought, because a bare path means nothing six months later: /audio/book.mp3 is on whichever machine happened to serve it that day. Idempotent on (project_id, content_hash), so the same manifest can be submitted twice, or by two operators who did not talk to each other, without the catalogue doubling - and a partly-present manifest reports what was new rather than just "done". An operator who submitted 300 entries and was told "done" would not know two were malformed and dropped, so malformed entries are named in the response instead. The listing is scoped by organization on every call. artifacts is platform-level, as every table a node serves from is, so a missing predicate there is a cross-tenant read rather than a style issue. A project in another tenant's schema is refused as *not found*, not as *not permitted*, because a refusal that says "this exists but is not yours" answers the question the caller was actually asking. Migration 0007 applied to the deployment. Mutation-verified: marking an ingest verified fails 1, dropping the idempotency check fails 2, and dropping the organization predicate from the listing fails 7. 704 passed, 4 skipped.
This commit is contained in:
@@ -762,6 +762,130 @@ class ComputeRepository:
|
||||
assert node is not None
|
||||
return node
|
||||
|
||||
# -- the catalogue -----------------------------------------------------
|
||||
|
||||
def register_artifacts(
|
||||
self,
|
||||
organization_id: UUID,
|
||||
project_id: UUID,
|
||||
specs: Sequence[Mapping[str, Any]],
|
||||
*,
|
||||
storage_node_id: UUID | None = None,
|
||||
actor: str | None = None,
|
||||
) -> dict[str, list[str]]:
|
||||
"""Declare files that already exist somewhere else.
|
||||
|
||||
The audio node is the master for the audiobooks, so this is how a library enters
|
||||
the system: the metadata needed to reason about a file, and a reference to where it
|
||||
is. No bytes are copied and none are assumed to be reachable - DECISION-6 is still
|
||||
unresolved, and this is what lets the system be useful before it is rather than
|
||||
after.
|
||||
|
||||
Idempotent on `(project_id, content_hash)`, so a manifest can be submitted twice,
|
||||
or by two operators who did not talk to each other, without the catalogue doubling.
|
||||
A file already present is counted as `skipped` rather than reported as new, because
|
||||
"42 of 300 added, 258 already registered" is the answer an operator needs and a
|
||||
silent success is not.
|
||||
|
||||
Every artifact lands `integrity = 'unverified'`, including when the reporting node
|
||||
supplied a perfectly good hash. A node reporting the hash of a file is a claim about
|
||||
those bytes; `verified_at` is set by reading them back, and nothing here reads them
|
||||
back, so claiming otherwise would make the column a decoration.
|
||||
"""
|
||||
|
||||
added: list[str] = []
|
||||
skipped: list[str] = []
|
||||
with self._connection() as connection, self._tx(connection):
|
||||
self._require_project_in_tenant(connection, organization_id, project_id)
|
||||
for spec in specs:
|
||||
content_hash = str(spec.get("content_hash") or "").strip()
|
||||
storage_uri = str(spec.get("storage_uri") or "").strip()
|
||||
if not content_hash or not storage_uri:
|
||||
# Counted as skipped, not silently dropped: a manifest entry without
|
||||
# both is malformed, and an operator who submitted 300 entries and got
|
||||
# "done" would not know two of them were ignored.
|
||||
skipped.append(str(spec.get("storage_key") or content_hash or "<unnamed>"))
|
||||
continue
|
||||
existing = connection.execute(
|
||||
"SELECT id FROM public.artifacts "
|
||||
"WHERE project_id = %s AND content_hash = %s",
|
||||
(project_id, content_hash),
|
||||
).fetchone()
|
||||
if existing is not None:
|
||||
skipped.append(str(existing["id"]))
|
||||
continue
|
||||
row = connection.execute(
|
||||
"""
|
||||
INSERT INTO public.artifacts (
|
||||
project_id, organization_id, lang, kind, status, content_hash,
|
||||
byte_size, media_type, storage_key, storage_uri, storage_node_id,
|
||||
integrity, verified_at, created_by_job_id, derivation
|
||||
)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'unverified', NULL,
|
||||
NULL, %s::jsonb)
|
||||
RETURNING id
|
||||
""",
|
||||
(
|
||||
project_id,
|
||||
organization_id,
|
||||
spec.get("lang"),
|
||||
str(spec.get("kind") or "unknown"),
|
||||
str(spec.get("status") or "intermediate"),
|
||||
content_hash,
|
||||
int(spec.get("byte_size") or 0),
|
||||
str(spec.get("media_type") or "application/octet-stream"),
|
||||
str(spec.get("storage_key") or storage_uri),
|
||||
storage_uri,
|
||||
storage_node_id,
|
||||
json.dumps(dict(spec.get("derivation") or {})),
|
||||
),
|
||||
).fetchone()
|
||||
added.append(str(row["id"]))
|
||||
|
||||
if added:
|
||||
self._node_event(
|
||||
connection,
|
||||
storage_node_id,
|
||||
"artifacts_registered",
|
||||
{
|
||||
"project_id": str(project_id),
|
||||
"added": len(added),
|
||||
"skipped": len(skipped),
|
||||
"actor": actor,
|
||||
},
|
||||
)
|
||||
return {"added": added, "skipped": skipped}
|
||||
|
||||
def list_artifacts(
|
||||
self,
|
||||
organization_id: UUID,
|
||||
*,
|
||||
project_id: UUID | None = None,
|
||||
kind: str | None = None,
|
||||
limit: int = 200,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""The catalogue, newest first.
|
||||
|
||||
Scoped by organization on every call. `artifacts` is a platform-level table and so
|
||||
is every other table a node serves from, which means a missing predicate here would
|
||||
be one organization's library readable by another - so the predicate is not
|
||||
optional and is not left to the caller to remember.
|
||||
"""
|
||||
|
||||
query = "SELECT * FROM public.artifacts WHERE organization_id = %s"
|
||||
params: list[Any] = [organization_id]
|
||||
if project_id is not None:
|
||||
query += " AND project_id = %s"
|
||||
params.append(project_id)
|
||||
if kind is not None:
|
||||
query += " AND kind = %s"
|
||||
params.append(kind)
|
||||
query += " ORDER BY created_at DESC LIMIT %s"
|
||||
params.append(limit)
|
||||
with self._connection() as connection:
|
||||
rows = connection.execute(query, params).fetchall()
|
||||
return [dict(row) for row in rows]
|
||||
|
||||
# -- assignments -------------------------------------------------------
|
||||
|
||||
def assign_job(
|
||||
@@ -1437,37 +1561,43 @@ class ComputeRepository:
|
||||
def _node_event(
|
||||
self,
|
||||
connection: Any,
|
||||
node_id: UUID,
|
||||
node_id: UUID | None,
|
||||
action: str,
|
||||
detail: Mapping[str, Any] | None = None,
|
||||
*,
|
||||
actor_user_id: UUID | None = None,
|
||||
) -> None:
|
||||
"""Record something that happened to a node, in the audit trail.
|
||||
"""Record something that happened, in the audit trail.
|
||||
|
||||
`job_events` is the wrong home for these: its `job_id` is not nullable, so a
|
||||
thing that happened to a *machine* - revoked credentials, a permanent removal -
|
||||
has no job to hang from. `audit_events` is nullable in every part and has a
|
||||
thing that happened to a *machine* - revoked credentials, a permanent removal - has
|
||||
no job to hang from. `audit_events` is nullable in every part and has a
|
||||
`target_type`/`target_id` pair, which is exactly this shape.
|
||||
|
||||
`node_id` may be None for an event that is about an organization rather than a
|
||||
machine - a library being registered by an operator, for instance. Recording a
|
||||
made-up identifier in `target_id` to satisfy the column would put a row in the audit
|
||||
trail pointing at a node that does not exist, which is worse than a null.
|
||||
|
||||
`actor_user_id` is nullable too, so an action taken by the controller on its own
|
||||
initiative is recorded as such rather than being attributed to whoever happened
|
||||
to be signed in.
|
||||
"""
|
||||
|
||||
node = self._node_row(connection, node_id)
|
||||
node = self._node_row(connection, node_id) if node_id is not None else None
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO public.audit_events (
|
||||
actor_user_id, organization_id, action, target_type, target_id, metadata
|
||||
)
|
||||
VALUES (%s, %s, %s, 'compute_node', %s, %s::jsonb)
|
||||
VALUES (%s, %s, %s, %s, %s, %s::jsonb)
|
||||
""",
|
||||
(
|
||||
actor_user_id,
|
||||
node["organization_id"] if node else None,
|
||||
action,
|
||||
str(node_id),
|
||||
"compute_node" if node_id is not None else "organization",
|
||||
str(node_id) if node_id is not None else None,
|
||||
json.dumps(dict(detail or {})),
|
||||
),
|
||||
)
|
||||
|
||||
51
umbratrans_web/migrations/0007_external_artifacts.sql
Normal file
51
umbratrans_web/migrations/0007_external_artifacts.sql
Normal file
@@ -0,0 +1,51 @@
|
||||
-- 0007: artifacts that live somewhere else.
|
||||
--
|
||||
-- The audio node is the master for the audiobooks. UmbraTrans holds the metadata needed
|
||||
-- to reason about a file and a reference to where it actually is; it does not hold the
|
||||
-- bytes. DECISION-6 (object storage) stays unresolved, and this is what lets the system
|
||||
-- be useful before it is resolved rather than after.
|
||||
--
|
||||
-- `storage_key` was already NOT NULL and already carried a path, so nothing here is
|
||||
-- required to register a file - these columns make the *reference* explicit and checkable
|
||||
-- rather than implicit in a string.
|
||||
|
||||
ALTER TABLE public.artifacts
|
||||
ADD COLUMN IF NOT EXISTS storage_uri text,
|
||||
ADD COLUMN IF NOT EXISTS storage_node_id uuid REFERENCES public.compute_nodes (id) ON DELETE SET NULL,
|
||||
ADD COLUMN IF NOT EXISTS integrity text NOT NULL DEFAULT 'unverified',
|
||||
ADD COLUMN IF NOT EXISTS verified_at timestamptz;
|
||||
|
||||
-- The audio node is the master, so the reference has to say which machine and how. A bare
|
||||
-- path means nothing to anybody reading this row six months later: `/audio/book.mp3` is
|
||||
-- on whichever node happened to serve it that day.
|
||||
COMMENT ON COLUMN public.artifacts.storage_uri IS
|
||||
'A URI naming where the bytes actually are, e.g. file:///media/audiobooks/… or '
|
||||
's3://…. Deliberately not assumed to be retrievable by this system: with no object '
|
||||
'storage behind it, nothing here fetches from it, and integrity is how a reader can '
|
||||
'tell a checked reference from an unchecked one.';
|
||||
|
||||
COMMENT ON COLUMN public.artifacts.integrity IS
|
||||
'unverified - a node reported it and nobody has checked the bytes against '
|
||||
'content_hash. verified - the bytes were read back and hashed, and verified_at says '
|
||||
'when. The default is unverified because a node reporting a hash is a claim, and '
|
||||
'treating the claim as the check is the mistake this column exists to prevent.';
|
||||
|
||||
-- Re-ingesting a library must not duplicate it. Content-addressed, so the natural key is
|
||||
-- the hash within a project: the same bytes under the same project are the same artifact
|
||||
-- however many times they are declared.
|
||||
--
|
||||
-- Partial on storage_uri rather than unique outright, so two projects may legitimately
|
||||
-- reference the same file - which is what happens when a shared library is referenced by
|
||||
-- more than one book - while one project cannot end up with two rows for one file.
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS artifacts_once_per_project
|
||||
ON public.artifacts (project_id, content_hash)
|
||||
WHERE storage_uri IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS artifacts_by_node
|
||||
ON public.artifacts (storage_node_id)
|
||||
WHERE storage_node_id IS NOT NULL;
|
||||
|
||||
-- Listing a project's catalogue is the only read the console and evaluation tooling do,
|
||||
-- and without this it is a sequential scan of every artifact ever ingested.
|
||||
CREATE INDEX IF NOT EXISTS artifacts_by_project
|
||||
ON public.artifacts (project_id, created_at);
|
||||
@@ -1046,6 +1046,146 @@ def fail_assignment(
|
||||
return _outcome_response(outcome)
|
||||
|
||||
|
||||
class ExternalArtifactSpec(RequestModel):
|
||||
"""One file that already exists somewhere, declared rather than uploaded.
|
||||
|
||||
Every bound mirrors a database constraint, for the same reason as `ArtifactSpec`:
|
||||
a value the INSERT rejects is a 500 a caller cannot act on.
|
||||
|
||||
``storage_uri`` is required and is not fetched. It says where the bytes are; nothing
|
||||
here goes and gets them, because there is no object storage to get them from
|
||||
(DECISION-6 is unresolved). What the row therefore records is that a file is claimed to
|
||||
exist at that address - see ``integrity``, which is ``unverified`` on every row written
|
||||
here and stays that way until something reads the bytes back and hashes them.
|
||||
"""
|
||||
|
||||
kind: str = Field(min_length=1, max_length=64)
|
||||
content_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
|
||||
storage_uri: str = Field(min_length=1, max_length=2048)
|
||||
storage_key: str | None = Field(default=None, max_length=1024)
|
||||
byte_size: int = Field(default=0, ge=0)
|
||||
media_type: str = Field(default="application/octet-stream", max_length=200)
|
||||
status: Literal["intermediate", "final", "superseded"] = "intermediate"
|
||||
lang: str | None = Field(default=None, max_length=16)
|
||||
derivation: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
|
||||
class RegisterArtifactsRequest(RequestModel):
|
||||
"""A manifest. One project's worth, declared in one call.
|
||||
|
||||
Bulk because a library is bulk: a finished audiobook is hundreds of files, and an
|
||||
operator submitting them one at a time would abandon it partway and leave a catalogue
|
||||
that is neither complete nor wrong - the worst state to be in.
|
||||
"""
|
||||
|
||||
project_id: UUID
|
||||
storage_node_id: UUID | None = None
|
||||
artifacts: list[ExternalArtifactSpec] = Field(min_length=1, max_length=2000)
|
||||
|
||||
|
||||
class RegisterArtifactsResponse(BaseModel):
|
||||
"""What was added and what was already there.
|
||||
|
||||
``skipped`` names the entries rather than counting them. An operator who submitted a
|
||||
300-file manifest and was told "done" would not know that two entries were malformed and
|
||||
silently dropped; being able to see which two is the difference between a catalogue you
|
||||
can rely on and one you cannot.
|
||||
"""
|
||||
|
||||
added: list[str]
|
||||
skipped: list[str]
|
||||
total_added: int
|
||||
|
||||
|
||||
class ArtifactResponse(BaseModel):
|
||||
id: UUID
|
||||
project_id: UUID
|
||||
organization_id: UUID
|
||||
kind: str
|
||||
status: str
|
||||
content_hash: str
|
||||
byte_size: int
|
||||
media_type: str
|
||||
storage_key: str
|
||||
storage_uri: str | None
|
||||
storage_node_id: UUID | None
|
||||
integrity: str
|
||||
verified_at: datetime | None
|
||||
lang: str | None
|
||||
created_at: datetime | None
|
||||
|
||||
|
||||
@admin_router.post("/artifacts", response_model=RegisterArtifactsResponse, status_code=201)
|
||||
def register_artifacts(
|
||||
payload: RegisterArtifactsRequest,
|
||||
request: Request,
|
||||
principal: Principal = Depends(require_admin),
|
||||
organization_id: UUID | None = Query(default=None),
|
||||
) -> RegisterArtifactsResponse:
|
||||
"""Declare files that already exist, against a project in this organization.
|
||||
|
||||
The audio node is the master for the audiobooks; this is how its library enters the
|
||||
system without the bytes moving. Idempotent on the file's hash within the project, so
|
||||
the same manifest can be submitted twice - or by two operators who did not talk to each
|
||||
other - without doubling the catalogue.
|
||||
"""
|
||||
|
||||
repository = compute_repository(request)
|
||||
target = organization_id or _organization_of(principal)
|
||||
outcome = repository.register_artifacts(
|
||||
target,
|
||||
payload.project_id,
|
||||
[spec.model_dump() for spec in payload.artifacts],
|
||||
storage_node_id=payload.storage_node_id,
|
||||
actor=getattr(principal, "email", None),
|
||||
)
|
||||
return RegisterArtifactsResponse(
|
||||
added=outcome["added"],
|
||||
skipped=outcome["skipped"],
|
||||
total_added=len(outcome["added"]),
|
||||
)
|
||||
|
||||
|
||||
@admin_router.get("/artifacts", response_model=list[ArtifactResponse])
|
||||
def list_artifacts(
|
||||
request: Request,
|
||||
principal: Principal = Depends(require_admin),
|
||||
organization_id: UUID | None = Query(default=None),
|
||||
project_id: UUID | None = Query(default=None),
|
||||
kind: str | None = Query(default=None),
|
||||
limit: int = Query(default=200, ge=1, le=1000),
|
||||
) -> list[ArtifactResponse]:
|
||||
"""The catalogue: what is claimed to exist, where, and whether anyone has checked.
|
||||
|
||||
Scoped to the caller's organization on every call. ``artifacts`` is platform-level, so
|
||||
a missing predicate here would be one tenant's library readable by another.
|
||||
"""
|
||||
|
||||
repository = compute_repository(request)
|
||||
target = organization_id or _organization_of(principal)
|
||||
rows = repository.list_artifacts(target, project_id=project_id, kind=kind, limit=limit)
|
||||
return [
|
||||
ArtifactResponse(
|
||||
id=row["id"],
|
||||
project_id=row["project_id"],
|
||||
organization_id=row["organization_id"],
|
||||
kind=row["kind"],
|
||||
status=row["status"],
|
||||
content_hash=row["content_hash"],
|
||||
byte_size=int(row["byte_size"]),
|
||||
media_type=row["media_type"],
|
||||
storage_key=row["storage_key"],
|
||||
storage_uri=row.get("storage_uri"),
|
||||
storage_node_id=row.get("storage_node_id"),
|
||||
integrity=row.get("integrity") or "unverified",
|
||||
verified_at=row.get("verified_at"),
|
||||
lang=row.get("lang"),
|
||||
created_at=row.get("created_at"),
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
@admin_router.get("/models", response_model=list[ModelResponse])
|
||||
def list_models(request: Request, principal: Principal = Depends(require_admin)) -> list[ModelResponse]:
|
||||
"""The global model registry.
|
||||
|
||||
324
umbratrans_web/tests/test_artifact_ingest.py
Normal file
324
umbratrans_web/tests/test_artifact_ingest.py
Normal file
@@ -0,0 +1,324 @@
|
||||
"""A library that already exists, declared rather than uploaded.
|
||||
|
||||
The audio node is the master for the audiobooks. UmbraTrans holds the metadata needed to
|
||||
reason about a file and a reference to where it is; it does not hold the bytes. DECISION-6
|
||||
(object storage) is still unresolved, and these are the tests for what works without it.
|
||||
|
||||
The property that matters most here is not the happy path. It is that a *reference* is not
|
||||
a *claim of retrievability*: a row that says a file exists at an address, produced by a node
|
||||
reporting it, is evidence that somebody said so - and the column that says so
|
||||
(`integrity`) starts at `unverified` and is only set by reading the bytes back. A design
|
||||
where ingesting a manifest marks the catalogue verified would make the first ingest the
|
||||
most reassuring and least true event in the system's life.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from umbratrans_web.compute import ComputeRepository
|
||||
|
||||
INTEGRATION_ENV = "UMBRATRANS_TEST_DSN"
|
||||
|
||||
|
||||
def _enabled() -> bool:
|
||||
return bool(os.environ.get(INTEGRATION_ENV)) and os.environ.get(
|
||||
"UMBRATRANS_TEST_DATABASE_IS_DISPOSABLE", ""
|
||||
) in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
integration = pytest.mark.skipif(
|
||||
not _enabled(),
|
||||
reason=f"needs {INTEGRATION_ENV} and an acknowledged disposable database",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def compute():
|
||||
"""A repository against the scratch database, emptied on teardown.
|
||||
|
||||
Copied rather than imported from `test_assignments.py` for the same reason that file
|
||||
copied it from `test_compute.py`: those fixtures are module-local on purpose, and
|
||||
hoisting them into `conftest.py` would make every file's fixtures depend on one file's
|
||||
imports.
|
||||
"""
|
||||
|
||||
from umbratrans_web.tests.test_compute import _require_scratch_database
|
||||
|
||||
_require_scratch_database()
|
||||
repository = ComputeRepository(
|
||||
os.environ[INTEGRATION_ENV], min_size=1, max_size=6, passwords=None
|
||||
)
|
||||
try:
|
||||
yield repository
|
||||
with repository._connection() as connection, connection.transaction(): # noqa: SLF001
|
||||
connection.execute(
|
||||
"TRUNCATE public.job_events, public.jobs, public.job_assignments, "
|
||||
"public.job_artifacts, public.artifacts, public.node_heartbeats, "
|
||||
"public.node_models, public.node_credentials, public.compute_devices, "
|
||||
"public.compute_nodes, public.model_registry CASCADE"
|
||||
)
|
||||
finally:
|
||||
repository.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tenant(compute):
|
||||
"""An organization and a project inside it, in the project's own schema.
|
||||
|
||||
Both are needed: `artifacts` is platform-level and carries an `organization_id`, while
|
||||
`projects` moved into a tenant schema in migration 0005 and has no such column. Getting
|
||||
that arrangement wrong is the failure that lets one tenant's library be listed against
|
||||
another's name.
|
||||
"""
|
||||
|
||||
from umbratrans_web.tenancy import tenant_search_path
|
||||
|
||||
organization = uuid4()
|
||||
project = uuid4()
|
||||
with compute._connection() as connection, compute._tx(connection):
|
||||
connection.execute(
|
||||
"INSERT INTO public.organizations (id, name, slug) VALUES (%s, %s, %s)",
|
||||
(organization, "Catalogue Test", f"catalogue-{organization.hex[:8]}"),
|
||||
)
|
||||
connection.execute(
|
||||
"SELECT set_config('search_path', %s, true)", (tenant_search_path(organization),)
|
||||
)
|
||||
connection.execute(
|
||||
"INSERT INTO projects (id, name, slug) VALUES (%s, %s, %s)",
|
||||
(project, "The Last Dwarf", f"last-dwarf-{project.hex[:8]}"),
|
||||
)
|
||||
return organization, project
|
||||
|
||||
|
||||
def _file(index: int, **overrides) -> dict:
|
||||
spec = {
|
||||
"kind": "audio",
|
||||
"content_hash": f"{index:064x}",
|
||||
"storage_uri": f"file:///media/audiobooks/last-dwarf/chapter-{index:03d}.mp3",
|
||||
"storage_key": f"chapter-{index:03d}.mp3",
|
||||
"byte_size": 1024 * index,
|
||||
"media_type": "audio/mpeg",
|
||||
"status": "final",
|
||||
}
|
||||
spec.update(overrides)
|
||||
return spec
|
||||
|
||||
|
||||
# -- registering a library ---------------------------------------------------
|
||||
|
||||
|
||||
@integration
|
||||
def test_a_file_that_exists_elsewhere_can_be_declared(compute, tenant) -> None:
|
||||
organization, project = tenant
|
||||
|
||||
outcome = compute.register_artifacts(organization, project, [_file(1)])
|
||||
|
||||
assert len(outcome["added"]) == 1
|
||||
listed = compute.list_artifacts(organization, project_id=project)
|
||||
assert len(listed) == 1
|
||||
row = listed[0]
|
||||
assert row["content_hash"] == f"{1:064x}"
|
||||
assert row["storage_uri"] == "file:///media/audiobooks/last-dwarf/chapter-001.mp3"
|
||||
|
||||
|
||||
@integration
|
||||
def test_a_declared_file_is_never_marked_verified(compute, tenant) -> None:
|
||||
"""The property this whole design rests on.
|
||||
|
||||
Ingesting is a claim. Something reading the bytes back and hashing them is a check, and
|
||||
nothing in this path reads them - there is no object storage to read them from. So a
|
||||
row that arrived `verified` would be claiming a check that has not happened.
|
||||
"""
|
||||
|
||||
organization, project = tenant
|
||||
compute.register_artifacts(organization, project, [_file(1)])
|
||||
|
||||
row = compute.list_artifacts(organization, project_id=project)[0]
|
||||
|
||||
assert row["integrity"] == "unverified", (
|
||||
"a declared artifact was recorded as verified, so the catalogue claims its "
|
||||
"contents were checked by something that never read them"
|
||||
)
|
||||
assert row["verified_at"] is None, "a verification time was invented"
|
||||
|
||||
|
||||
@integration
|
||||
def test_a_declared_file_says_which_node_holds_it(compute, tenant) -> None:
|
||||
"""A bare path means nothing six months later.
|
||||
|
||||
`/audio/book.mp3` is on whichever machine happened to serve it that day, so the node is
|
||||
part of the reference rather than an afterthought.
|
||||
"""
|
||||
|
||||
organization, project = tenant
|
||||
node = uuid4()
|
||||
with compute._connection() as connection, compute._tx(connection):
|
||||
connection.execute(
|
||||
"INSERT INTO public.compute_nodes (id, organization_id, name, status) "
|
||||
"VALUES (%s, %s, 'library-holder', 'online')",
|
||||
(node, organization),
|
||||
)
|
||||
|
||||
compute.register_artifacts(organization, project, [_file(1)], storage_node_id=node)
|
||||
|
||||
row = compute.list_artifacts(organization, project_id=project)[0]
|
||||
assert row["storage_node_id"] == node, (
|
||||
"the catalog does not say which machine holds the file"
|
||||
)
|
||||
|
||||
|
||||
# -- idempotency -------------------------------------------------------------
|
||||
|
||||
|
||||
@integration
|
||||
def test_registering_the_same_library_twice_does_not_double_it(compute, tenant) -> None:
|
||||
"""A manifest may be submitted twice, or by two operators who did not talk.
|
||||
|
||||
Content-addressed, so the same bytes under the same project are the same artifact. A
|
||||
catalogue that doubles on every import is one nobody can answer a question against.
|
||||
"""
|
||||
|
||||
organization, project = tenant
|
||||
first = compute.register_artifacts(organization, project, [_file(1), _file(2)])
|
||||
second = compute.register_artifacts(organization, project, [_file(1), _file(2)])
|
||||
|
||||
assert len(first["added"]) == 2
|
||||
assert second["added"] == []
|
||||
assert len(second["skipped"]) == 2
|
||||
assert len(compute.list_artifacts(organization, project_id=project)) == 2
|
||||
|
||||
|
||||
@integration
|
||||
def test_a_manifest_partly_present_reports_what_was_new(compute, tenant) -> None:
|
||||
"""The answer an operator needs is "42 of 300 added, 258 already there"."""
|
||||
|
||||
organization, project = tenant
|
||||
compute.register_artifacts(organization, project, [_file(1)])
|
||||
|
||||
outcome = compute.register_artifacts(
|
||||
organization, project, [_file(1), _file(2), _file(3)]
|
||||
)
|
||||
|
||||
assert len(outcome["added"]) == 2
|
||||
assert len(outcome["skipped"]) == 1
|
||||
|
||||
|
||||
# -- malformed entries -------------------------------------------------------
|
||||
|
||||
|
||||
@integration
|
||||
def test_a_malformed_entry_is_named_rather_than_silently_dropped(compute, tenant) -> None:
|
||||
"""An operator who submitted 300 entries and was told "done" would not know.
|
||||
|
||||
Two of them missing a hash is a catalogue that is neither complete nor obviously
|
||||
wrong - the state that is hardest to notice and hardest to trust.
|
||||
"""
|
||||
|
||||
organization, project = tenant
|
||||
|
||||
outcome = compute.register_artifacts(
|
||||
organization,
|
||||
project,
|
||||
[
|
||||
_file(1),
|
||||
{"kind": "audio", "storage_uri": "file:///media/x.mp3"},
|
||||
{"kind": "audio", "content_hash": f"{9:064x}"},
|
||||
],
|
||||
)
|
||||
|
||||
assert len(outcome["added"]) == 1
|
||||
assert len(outcome["skipped"]) == 2, "malformed entries were dropped without being named"
|
||||
assert len(compute.list_artifacts(organization, project_id=project)) == 1
|
||||
|
||||
|
||||
# -- tenancy -----------------------------------------------------------------
|
||||
|
||||
|
||||
@integration
|
||||
def test_one_organizations_catalogue_is_not_another(compute, tenant) -> None:
|
||||
"""`artifacts` is platform-level, so the predicate is the only boundary.
|
||||
|
||||
Every table a node serves from is platform-level by design - a node is a machine, not a
|
||||
tenant. That makes a missing organization predicate in a listing a cross-tenant read,
|
||||
which is the whole reason this test exists rather than a comment.
|
||||
"""
|
||||
|
||||
organization, project = tenant
|
||||
other = uuid4()
|
||||
with compute._connection() as connection, compute._tx(connection):
|
||||
connection.execute(
|
||||
"INSERT INTO public.organizations (id, name, slug) VALUES (%s, %s, %s)",
|
||||
(other, "Other Tenant", f"other-{other.hex[:8]}"),
|
||||
)
|
||||
|
||||
compute.register_artifacts(organization, project, [_file(1)])
|
||||
|
||||
assert len(compute.list_artifacts(organization)) == 1
|
||||
assert compute.list_artifacts(other) == [], (
|
||||
"another organization can list this catalog"
|
||||
)
|
||||
|
||||
|
||||
@integration
|
||||
def test_a_project_in_another_tenants_schema_is_refused(compute, tenant) -> None:
|
||||
"""Naming somebody else's project must not put their files in your catalogue.
|
||||
|
||||
Refused as *not found* rather than *not permitted*, and that distinction is the point:
|
||||
a refusal that says "this exists but is not yours" answers the question the caller was
|
||||
asking, which is whether that project id is real. So the test asserts the message
|
||||
carries nothing beyond the id the caller already supplied.
|
||||
"""
|
||||
|
||||
from umbratrans_web.errors import NotFoundError
|
||||
from umbratrans_web.tenancy import tenant_search_path
|
||||
|
||||
organization, _project = tenant
|
||||
theirs = uuid4()
|
||||
with compute._connection() as connection, compute._tx(connection):
|
||||
connection.execute(
|
||||
"INSERT INTO public.organizations (id, name, slug) VALUES (%s, %s, %s)",
|
||||
(theirs, "Someone Else", f"else-{theirs.hex[:8]}"),
|
||||
)
|
||||
connection.execute(
|
||||
"SELECT set_config('search_path', %s, true)", (tenant_search_path(theirs),)
|
||||
)
|
||||
connection.execute(
|
||||
"INSERT INTO projects (id, name, slug) VALUES (%s, %s, %s)",
|
||||
(theirs, "Their Book", f"their-book-{theirs.hex[:8]}"),
|
||||
)
|
||||
|
||||
with pytest.raises(NotFoundError) as error:
|
||||
compute.register_artifacts(organization, theirs, [_file(1)])
|
||||
|
||||
message = str(error.value)
|
||||
assert "Someone Else" not in message, (
|
||||
f"the refusal names the other organization: {message}"
|
||||
)
|
||||
assert "not yours" not in message.lower() and "forbidden" not in message.lower(), (
|
||||
f"the refusal confirms the project exists: {message}"
|
||||
)
|
||||
|
||||
|
||||
@integration
|
||||
def test_a_whole_library_can_be_registered_in_one_call(compute, tenant) -> None:
|
||||
"""Because a finished audiobook is hundreds of files.
|
||||
|
||||
An operator submitting them one at a time would abandon it partway and leave a
|
||||
catalogue that is neither complete nor wrong.
|
||||
"""
|
||||
|
||||
organization, project = tenant
|
||||
manifest = [_file(index) for index in range(1, 201)]
|
||||
|
||||
outcome = compute.register_artifacts(organization, project, manifest)
|
||||
|
||||
assert len(outcome["added"]) == 200
|
||||
assert len(compute.list_artifacts(organization, project_id=project, limit=500)) == 200
|
||||
|
||||
|
||||
def _unused_import_guard() -> type: # pragma: no cover
|
||||
return ComputeRepository
|
||||
Reference in New Issue
Block a user