gabriel / musehub public
factories.py python
310 lines 9.0 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """Test factories for MuseHub ORM models.
2
3 Provides two layers:
4 1. ``*Factory`` classes (factory_boy ``Factory`` subclasses) that generate
5 realistic attribute dictionaries without touching the database.
6 2. Async ``create_*`` helpers that instantiate the ORM model from the
7 factory data, persist it, and return the refreshed ORM object.
8
9 Usage in tests::
10
11 from tests.factories import create_repo, create_profile, RepoFactory
12
13 async def test_something(db_session):
14 repo = await create_repo(db_session, owner="alice", visibility="public")
15 assert repo.owner == "alice"
16
17 # Data-only (no DB) — useful for unit-testing pure functions:
18 data = RepoFactory(name="My Jazz EP", owner="charlie")
19 assert data["slug"] == "my-jazz-ep"
20 """
21 from __future__ import annotations
22
23 import hashlib
24 import re
25 import uuid
26 from datetime import datetime, timezone
27
28 import factory
29 from sqlalchemy.ext.asyncio import AsyncSession
30
31 from musehub.db import musehub_models as db
32 from musehub.types.json_types import JSONValue
33
34
35 # ---------------------------------------------------------------------------
36 # Helpers
37 # ---------------------------------------------------------------------------
38
39 def _uid() -> str:
40 return str(uuid.uuid4())
41
42
43 def _now() -> datetime:
44 return datetime.now(tz=timezone.utc)
45
46
47 def _slugify(name: str) -> str:
48 """Convert a human-readable name to a URL-safe slug."""
49 return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-") or "repo"
50
51
52 def _sha(seed: str) -> str:
53 return hashlib.sha256(seed.encode()).hexdigest()
54
55
56 # ---------------------------------------------------------------------------
57 # Attribute factories (no DB access)
58 # ---------------------------------------------------------------------------
59
60 class RepoFactory(factory.Factory):
61 """Generate attribute dicts for MusehubRepo."""
62
63 class Meta:
64 model = dict
65
66 name: str = factory.Sequence(lambda n: f"Test Repo {n}")
67 owner: str = "testuser"
68 slug: str = factory.LazyAttribute(lambda o: _slugify(o.name))
69 visibility: str = "public"
70 owner_user_id: str = factory.LazyFunction(_uid)
71 description: str = factory.LazyAttribute(lambda o: f"Description for {o.name}")
72 tags = factory.LazyFunction(list)
73
74
75 class BranchFactory(factory.Factory):
76 class Meta:
77 model = dict
78
79 name: str = "main"
80 head_commit_id: str | None = None
81
82
83 class CommitFactory(factory.Factory):
84 class Meta:
85 model = dict
86
87 commit_id: str = factory.LazyFunction(lambda: _sha(str(uuid.uuid4())))
88 message: str = factory.Sequence(lambda n: f"feat: commit number {n}")
89 author: str = "testuser"
90 branch: str = "main"
91 parent_ids = factory.LazyFunction(list)
92 snapshot_id: str | None = None
93 timestamp: datetime = factory.LazyFunction(_now)
94
95
96 class ProfileFactory(factory.Factory):
97 class Meta:
98 model = dict
99
100 user_id: str = factory.LazyFunction(_uid)
101 username: str = factory.Sequence(lambda n: f"user{n}")
102 display_name: str = factory.LazyAttribute(lambda o: o.username.title())
103 bio: str = "A musician who uses Muse VCS."
104 avatar_url: str | None = None
105 location: str | None = None
106 website_url: str | None = None
107 social_url: str | None = None
108 is_verified: bool = False
109 cc_license: str | None = None
110 pinned_repo_ids = factory.LazyFunction(list)
111
112
113 class IssueFactory(factory.Factory):
114 class Meta:
115 model = dict
116
117 title: str = factory.Sequence(lambda n: f"Issue #{n}")
118 body: str = "Issue body text."
119 author: str = "testuser"
120 status: str = "open"
121
122
123 class SessionFactory(factory.Factory):
124 class Meta:
125 model = dict
126
127 session_id: str = factory.LazyFunction(_uid)
128 participants = factory.LazyFunction(lambda: ["testuser"])
129 commits = factory.LazyFunction(list)
130 notes: str | None = None
131 location: str | None = None
132 intent: str | None = None
133
134
135 # ---------------------------------------------------------------------------
136 # Async persistence helpers
137 # ---------------------------------------------------------------------------
138
139 async def create_repo(
140 session: AsyncSession,
141 **kwargs: JSONValue,
142 ) -> db.MusehubRepo:
143 """Insert and return a MusehubRepo row using RepoFactory defaults."""
144 from musehub.core.genesis import compute_repo_id
145 data = RepoFactory(**kwargs)
146 created_at = _now()
147 owner_user_id = str(data["owner_user_id"])
148 slug = str(data["slug"])
149 domain = str(data.get("domain_id") or "")
150 repo_id = compute_repo_id(owner_user_id, slug, domain, created_at.isoformat())
151 repo = db.MusehubRepo(
152 repo_id=repo_id,
153 name=data["name"],
154 owner=data["owner"],
155 slug=slug,
156 visibility=data["visibility"],
157 owner_user_id=owner_user_id,
158 description=data["description"],
159 tags=data["tags"],
160 created_at=created_at,
161 )
162 session.add(repo)
163 await session.commit()
164 await session.refresh(repo)
165 return repo
166
167
168 async def create_branch(
169 session: AsyncSession,
170 repo_id: str,
171 **kwargs: JSONValue,
172 ) -> db.MusehubBranch:
173 """Insert and return a MusehubBranch row."""
174 from musehub.core.genesis import compute_branch_id
175 data = BranchFactory(**kwargs)
176 name = str(data["name"])
177 branch = db.MusehubBranch(
178 branch_id=compute_branch_id(repo_id, name),
179 repo_id=repo_id,
180 name=name,
181 head_commit_id=data.get("head_commit_id"),
182 )
183 session.add(branch)
184 await session.commit()
185 await session.refresh(branch)
186 return branch
187
188
189 async def create_commit(
190 session: AsyncSession,
191 repo_id: str,
192 **kwargs: JSONValue,
193 ) -> db.MusehubCommit:
194 """Insert and return a MusehubCommit row."""
195 data = CommitFactory(**kwargs)
196 commit = db.MusehubCommit(
197 commit_id=data["commit_id"],
198 repo_id=repo_id,
199 message=data["message"],
200 author=data["author"],
201 branch=data["branch"],
202 parent_ids=data["parent_ids"],
203 snapshot_id=data.get("snapshot_id"),
204 timestamp=data.get("timestamp") or _now(),
205 )
206 session.add(commit)
207 await session.commit()
208 await session.refresh(commit)
209 return commit
210
211
212 async def create_profile(
213 session: AsyncSession,
214 **kwargs: JSONValue,
215 ) -> db.MusehubIdentity:
216 """Insert and return a MusehubIdentity row."""
217 data = ProfileFactory(**kwargs)
218 profile = db.MusehubIdentity(
219 identity_id=data["user_id"],
220 handle=data["username"],
221 identity_type="human",
222 display_name=data["display_name"],
223 bio=data["bio"],
224 avatar_url=data.get("avatar_url"),
225 location=data.get("location"),
226 website_url=data.get("website_url"),
227 social_url=data.get("social_url"),
228 is_verified=data["is_verified"],
229 cc_license=data.get("cc_license"),
230 )
231 session.add(profile)
232 await session.commit()
233 await session.refresh(profile)
234 return profile
235
236
237 async def create_repo_with_branch(
238 session: AsyncSession,
239 **kwargs: JSONValue,
240 ) -> tuple[db.MusehubRepo, db.MusehubBranch]:
241 """Convenience: create a repo + default 'main' branch atomically."""
242 repo = await create_repo(session, **kwargs)
243 branch = await create_branch(session, repo_id=str(repo.repo_id), name="main")
244 return repo, branch
245
246
247 async def create_issue(
248 session: AsyncSession,
249 repo_id: str,
250 *,
251 author: str = "testuser",
252 title: str = "Test issue",
253 body: str = "",
254 state: str = "open",
255 number: int | None = None,
256 ) -> db.MusehubIssue:
257 """Insert and return a MusehubIssue row."""
258 from sqlalchemy import func, select as sa_select
259 if number is None:
260 result = await session.execute(
261 sa_select(func.count()).select_from(db.MusehubIssue).where(db.MusehubIssue.repo_id == repo_id)
262 )
263 number = (result.scalar() or 0) + 1
264 issue = db.MusehubIssue(
265 repo_id=repo_id,
266 number=number,
267 title=title,
268 body=body,
269 state=state,
270 author=author,
271 )
272 session.add(issue)
273 await session.commit()
274 await session.refresh(issue)
275 return issue
276
277
278 async def create_proposal(
279 session: AsyncSession,
280 repo_id: str,
281 *,
282 author: str = "testuser",
283 title: str = "Test proposal",
284 body: str = "",
285 state: str = "open",
286 from_branch: str = "feature/test",
287 to_branch: str = "main",
288 proposal_number: int | None = None,
289 ) -> db.MusehubProposal:
290 """Insert and return a MusehubProposal row."""
291 from sqlalchemy import func, select as sa_select
292 if proposal_number is None:
293 result = await session.execute(
294 sa_select(func.count()).select_from(db.MusehubProposal).where(db.MusehubProposal.repo_id == repo_id)
295 )
296 proposal_number = (result.scalar() or 0) + 1
297 proposal = db.MusehubProposal(
298 repo_id=repo_id,
299 proposal_number=proposal_number,
300 title=title,
301 body=body,
302 state=state,
303 author=author,
304 from_branch=from_branch,
305 to_branch=to_branch,
306 )
307 session.add(proposal)
308 await session.commit()
309 await session.refresh(proposal)
310 return proposal
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago