gabriel / musehub public
factories.py python
299 lines 8.5 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 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.muse_contracts.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 twitter_handle: 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 data = RepoFactory(**kwargs)
145 repo = db.MusehubRepo(
146 name=data["name"],
147 owner=data["owner"],
148 slug=data["slug"],
149 visibility=data["visibility"],
150 owner_user_id=data["owner_user_id"],
151 description=data["description"],
152 tags=data["tags"],
153 )
154 session.add(repo)
155 await session.commit()
156 await session.refresh(repo)
157 return repo
158
159
160 async def create_branch(
161 session: AsyncSession,
162 repo_id: str,
163 **kwargs: JSONValue,
164 ) -> db.MusehubBranch:
165 """Insert and return a MusehubBranch row."""
166 data = BranchFactory(**kwargs)
167 branch = db.MusehubBranch(
168 repo_id=repo_id,
169 name=data["name"],
170 head_commit_id=data.get("head_commit_id"),
171 )
172 session.add(branch)
173 await session.commit()
174 await session.refresh(branch)
175 return branch
176
177
178 async def create_commit(
179 session: AsyncSession,
180 repo_id: str,
181 **kwargs: JSONValue,
182 ) -> db.MusehubCommit:
183 """Insert and return a MusehubCommit row."""
184 data = CommitFactory(**kwargs)
185 commit = db.MusehubCommit(
186 commit_id=data["commit_id"],
187 repo_id=repo_id,
188 message=data["message"],
189 author=data["author"],
190 branch=data["branch"],
191 parent_ids=data["parent_ids"],
192 snapshot_id=data.get("snapshot_id"),
193 timestamp=data.get("timestamp") or _now(),
194 )
195 session.add(commit)
196 await session.commit()
197 await session.refresh(commit)
198 return commit
199
200
201 async def create_profile(
202 session: AsyncSession,
203 **kwargs: JSONValue,
204 ) -> db.MusehubIdentity:
205 """Insert and return a MusehubIdentity row."""
206 data = ProfileFactory(**kwargs)
207 profile = db.MusehubIdentity(
208 id=data["user_id"],
209 handle=data["username"],
210 identity_type="human",
211 display_name=data["display_name"],
212 bio=data["bio"],
213 avatar_url=data.get("avatar_url"),
214 location=data.get("location"),
215 website_url=data.get("website_url"),
216 twitter_handle=data.get("twitter_handle"),
217 is_verified=data["is_verified"],
218 cc_license=data.get("cc_license"),
219 )
220 session.add(profile)
221 await session.commit()
222 await session.refresh(profile)
223 return profile
224
225
226 async def create_repo_with_branch(
227 session: AsyncSession,
228 **kwargs: JSONValue,
229 ) -> tuple[db.MusehubRepo, db.MusehubBranch]:
230 """Convenience: create a repo + default 'main' branch atomically."""
231 repo = await create_repo(session, **kwargs)
232 branch = await create_branch(session, repo_id=str(repo.repo_id), name="main")
233 return repo, branch
234
235
236 async def create_issue(
237 session: AsyncSession,
238 repo_id: str,
239 *,
240 author: str = "testuser",
241 title: str = "Test issue",
242 body: str = "",
243 state: str = "open",
244 number: int | None = None,
245 ) -> db.MusehubIssue:
246 """Insert and return a MusehubIssue row."""
247 from sqlalchemy import func, select as sa_select
248 if number is None:
249 result = await session.execute(
250 sa_select(func.count()).select_from(db.MusehubIssue).where(db.MusehubIssue.repo_id == repo_id)
251 )
252 number = (result.scalar() or 0) + 1
253 issue = db.MusehubIssue(
254 repo_id=repo_id,
255 number=number,
256 title=title,
257 body=body,
258 state=state,
259 author=author,
260 )
261 session.add(issue)
262 await session.commit()
263 await session.refresh(issue)
264 return issue
265
266
267 async def create_proposal(
268 session: AsyncSession,
269 repo_id: str,
270 *,
271 author: str = "testuser",
272 title: str = "Test proposal",
273 body: str = "",
274 state: str = "open",
275 from_branch: str = "feature/test",
276 to_branch: str = "main",
277 proposal_number: int | None = None,
278 ) -> db.MusehubProposal:
279 """Insert and return a MusehubProposal row."""
280 from sqlalchemy import func, select as sa_select
281 if proposal_number is None:
282 result = await session.execute(
283 sa_select(func.count()).select_from(db.MusehubProposal).where(db.MusehubProposal.repo_id == repo_id)
284 )
285 proposal_number = (result.scalar() or 0) + 1
286 proposal = db.MusehubProposal(
287 repo_id=repo_id,
288 proposal_number=proposal_number,
289 title=title,
290 body=body,
291 state=state,
292 author=author,
293 from_branch=from_branch,
294 to_branch=to_branch,
295 )
296 session.add(proposal)
297 await session.commit()
298 await session.refresh(proposal)
299 return proposal
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago