Ben Stull
140 lines
6.4 KiB
Python
140 lines
6.4 KiB
Python
"""Migration 029 — the collection grain beneath projects (§22 three-tier S1).
|
|
|
|
Proves: a `collections` table exists with one default collection per project
|
|
(id='default', subfolder=repo root); the per-corpus fields (type, initial_state)
|
|
moved off `projects`; the 13 entry-corpus tables re-key (project_id,slug) ->
|
|
(collection_id,slug) with the composite PK/FK enforced; and project_members
|
|
generalises into memberships(scope_type, …) with the role enum collapsed.
|
|
Template: test_migration_028_project_scoped_keys.py.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app import db
|
|
|
|
|
|
class _Cfg:
|
|
def __init__(self, path):
|
|
self.database_path = path
|
|
|
|
|
|
def _fresh_db():
|
|
d = tempfile.mkdtemp()
|
|
path = Path(d) / "t.db"
|
|
db.run_migrations(_Cfg(str(path)))
|
|
return db.connect(str(path))
|
|
|
|
|
|
def test_collections_table_exists_with_default_per_project():
|
|
conn = _fresh_db()
|
|
cols = {r["name"] for r in conn.execute("PRAGMA table_info(collections)")}
|
|
assert {"id", "project_id", "type", "subfolder",
|
|
"initial_state", "visibility", "name", "registry_sha"} <= cols
|
|
# one default collection seeded for the bootstrap 'default' project (026)
|
|
row = conn.execute(
|
|
"SELECT id, project_id, subfolder FROM collections WHERE project_id='default'"
|
|
).fetchone()
|
|
assert row is not None
|
|
assert row["id"] == "default"
|
|
assert row["subfolder"] == "" # repo root
|
|
|
|
|
|
def test_per_corpus_fields_moved_off_projects():
|
|
conn = _fresh_db()
|
|
proj_cols = {r["name"] for r in conn.execute("PRAGMA table_info(projects)")}
|
|
assert "type" not in proj_cols
|
|
assert "initial_state" not in proj_cols
|
|
# projects keeps the grouping-tier fields
|
|
assert {"id", "name", "content_repo", "visibility"} <= proj_cols
|
|
|
|
|
|
def test_entry_tables_rekeyed_to_collection_id():
|
|
conn = _fresh_db()
|
|
for t in ("cached_rfcs", "cached_branches", "stars", "watches",
|
|
"rfc_collaborators", "contribution_requests", "proposed_use_cases",
|
|
"branch_visibility", "branch_contribute_grants", "pr_seen",
|
|
"branch_chat_seen", "funder_consents", "rfc_invitations"):
|
|
cols = {r["name"] for r in conn.execute(f"PRAGMA table_info({t})")}
|
|
assert "collection_id" in cols, f"{t} missing collection_id"
|
|
assert "project_id" not in cols, f"{t} still has project_id"
|
|
|
|
|
|
def test_cached_rfcs_pk_is_collection_slug():
|
|
conn = _fresh_db()
|
|
# a second collection under the default project
|
|
conn.execute(
|
|
"INSERT INTO collections (id, project_id, type, subfolder, initial_state, visibility, name) "
|
|
"VALUES ('c2','default','document','specs','active','public','Specs')"
|
|
)
|
|
conn.execute("INSERT INTO cached_rfcs (slug, title, state, collection_id) VALUES ('intro','A','active','default')")
|
|
conn.execute("INSERT INTO cached_rfcs (slug, title, state, collection_id) VALUES ('intro','B','active','c2')")
|
|
n = conn.execute("SELECT COUNT(*) c FROM cached_rfcs WHERE slug='intro'").fetchone()["c"]
|
|
assert n == 2
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute("INSERT INTO cached_rfcs (slug, title, state, collection_id) VALUES ('intro','dup','active','default')")
|
|
|
|
|
|
def test_cached_rfcs_collection_fk_enforced():
|
|
conn = _fresh_db()
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute("INSERT INTO cached_rfcs (slug, title, state, collection_id) VALUES ('x','X','active','nope')")
|
|
|
|
|
|
def test_collaborator_fk_is_composite_on_collection():
|
|
conn = _fresh_db()
|
|
conn.execute(
|
|
"INSERT INTO collections (id, project_id, type, subfolder, initial_state, visibility, name) "
|
|
"VALUES ('c2','default','document','specs','active','public','Specs')"
|
|
)
|
|
conn.execute("INSERT INTO cached_rfcs (slug, title, state, collection_id) VALUES ('intro','A','active','c2')")
|
|
conn.execute("INSERT INTO users (id, gitea_login, display_name, role) VALUES (1,'a','A','contributor')")
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
conn.execute(
|
|
"INSERT INTO rfc_collaborators (rfc_slug, user_id, role_in_rfc, collection_id) "
|
|
"VALUES ('intro',1,'contributor','c2')"
|
|
)
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
# same slug, a collection with no such entry — composite FK rejects
|
|
conn.execute(
|
|
"INSERT INTO rfc_collaborators (rfc_slug, user_id, role_in_rfc, collection_id) "
|
|
"VALUES ('intro',1,'contributor','default')"
|
|
)
|
|
|
|
|
|
def test_stars_unique_now_scoped_by_collection():
|
|
conn = _fresh_db()
|
|
conn.execute(
|
|
"INSERT INTO collections (id, project_id, type, subfolder, initial_state, visibility, name) "
|
|
"VALUES ('c2','default','document','specs','active','public','Specs')"
|
|
)
|
|
conn.execute("INSERT INTO users (id, gitea_login, display_name, role) VALUES (1,'a','A','contributor')")
|
|
conn.execute("INSERT INTO cached_rfcs (slug, title, state, collection_id) VALUES ('intro','A','active','default')")
|
|
conn.execute("INSERT INTO cached_rfcs (slug, title, state, collection_id) VALUES ('intro','B','active','c2')")
|
|
conn.execute("INSERT INTO stars (user_id, rfc_slug, collection_id) VALUES (1,'intro','default')")
|
|
conn.execute("INSERT INTO stars (user_id, rfc_slug, collection_id) VALUES (1,'intro','c2')")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute("INSERT INTO stars (user_id, rfc_slug, collection_id) VALUES (1,'intro','default')")
|
|
|
|
|
|
def test_memberships_table_replaces_project_members():
|
|
conn = _fresh_db()
|
|
cols = {r["name"] for r in conn.execute("PRAGMA table_info(memberships)")}
|
|
assert {"scope_type", "scope_id", "user_id", "role", "granted_by", "granted_at"} <= cols
|
|
# project_members is gone
|
|
assert conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='project_members'"
|
|
).fetchone() is None
|
|
conn.execute("INSERT INTO users (id, gitea_login, display_name, role) VALUES (9,'x','X','contributor')")
|
|
conn.execute("INSERT INTO memberships (scope_type, scope_id, user_id, role) VALUES ('project','default',9,'owner')")
|
|
# scope_type and role are CHECK-constrained
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute("INSERT INTO memberships (scope_type, scope_id, user_id, role) VALUES ('bogus','default',9,'owner')")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute("INSERT INTO memberships (scope_type, scope_id, user_id, role) VALUES ('project','default',9,'viewer')")
|