"""ORM model instances must have correct Python-side attribute values at construction time — no DB flush or refresh required. This guards against the class of bug where service code calls a Pydantic serialiser (e.g. _to_repo_response) on a freshly constructed ORM object and receives None for columns that have a server_default or column-level default, because SQLAlchemy's ``default=`` only fires at INSERT time in non-dataclass models. """ from __future__ import annotations from datetime import datetime, timezone from muse.core.types import fake_id from musehub.db.musehub_auth_models import MusehubAuthKey from musehub.db.musehub_collaborator_models import MusehubCollaborator from musehub.db.musehub_identity_models import MusehubIdentity from musehub.db.musehub_intel_models import MusehubSymbolIntel from musehub.db.musehub_release_models import MusehubRelease from musehub.db.musehub_repo_models import ( MusehubBranch, MusehubCommit, MusehubMist, MusehubRepo, MusehubSession, MusehubSnapshot, ) from musehub.db.musehub_social_models import ( MusehubIssue, MusehubIssueEvent, MusehubProposal, MusehubProposalComment, ) from musehub.db.musehub_webhook_models import MusehubWebhook _REPO_ID = fake_id("repo") _OWNER_ID = fake_id("owner") _BRANCH_ID = fake_id("branch") _COMMIT_ID = fake_id("commit") _SNAP_ID = fake_id("snapshot") _IDENTITY_ID = fake_id("identity") _ISSUE_ID = fake_id("issue") _PROPOSAL_ID = fake_id("proposal") _KEY_ID = fake_id("key") _COLLAB_ID = fake_id("collab") def test_musehub_repo_scalar_defaults_at_construction() -> None: repo = MusehubRepo( repo_id=_REPO_ID, name="test", owner="gabriel", slug="test", owner_user_id=_OWNER_ID, ) assert repo.default_branch == "main" assert repo.visibility == "public" assert repo.description == "" assert repo.domain_id == "code" assert repo.training_opt_out is False assert repo.settings is None assert repo.pushed_at is None def test_musehub_repo_mutable_defaults_are_isolated() -> None: """Each instance must get its own list/dict, not a shared mutable object.""" r1 = MusehubRepo(repo_id=_REPO_ID, name="t", owner="x", slug="t", owner_user_id=_OWNER_ID) r2 = MusehubRepo(repo_id=_REPO_ID, name="t", owner="x", slug="t", owner_user_id=_OWNER_ID) assert r1.tags == [] assert r1.domain_meta == {} r1.tags.append("jazz") assert r2.tags == [], "mutable default must not be shared between instances" def test_musehub_repo_timestamps_set_at_construction() -> None: repo = MusehubRepo( repo_id=_REPO_ID, name="t", owner="x", slug="t", owner_user_id=_OWNER_ID, ) assert isinstance(repo.created_at, datetime) assert isinstance(repo.updated_at, datetime) def test_musehub_snapshot_defaults_at_construction() -> None: snap = MusehubSnapshot( snapshot_id=_SNAP_ID, manifest_blob=b"\x80", ) assert snap.directories == [] assert snap.entry_count == 0 assert isinstance(snap.created_at, datetime) def test_musehub_identity_defaults_at_construction() -> None: identity = MusehubIdentity( identity_id=_IDENTITY_ID, handle="gabriel", ) assert identity.identity_type == "human" assert identity.agent_capabilities == [] assert identity.is_verified is False assert identity.pinned_repo_ids == [] assert isinstance(identity.created_at, datetime) assert isinstance(identity.updated_at, datetime) def test_musehub_issue_defaults_at_construction() -> None: issue = MusehubIssue( issue_id=_ISSUE_ID, repo_id=_REPO_ID, number=1, title="bug: something broken", ) assert issue.body == "" assert issue.state == "open" assert issue.labels == [] assert issue.symbol_anchors == [] assert issue.commit_anchors == [] assert issue.author == "" assert issue.assignee is None assert issue.agent_id == "" assert issue.model_id == "" assert isinstance(issue.created_at, datetime) assert isinstance(issue.updated_at, datetime) def test_musehub_proposal_defaults_at_construction() -> None: proposal = MusehubProposal( proposal_id=_PROPOSAL_ID, repo_id=_REPO_ID, proposal_number=1, title="feat: new thing", from_branch="feat/new-thing", to_branch="main", ) assert proposal.body == "" assert proposal.state == "open" assert proposal.merge_commit_id is None assert proposal.merged_at is None assert proposal.author == "" assert isinstance(proposal.created_at, datetime) assert isinstance(proposal.updated_at, datetime) assert proposal.domain_diff is None assert proposal.risk_score is None assert proposal.blast_delta is None assert proposal.breakage_count == 0 assert proposal.test_gap_count == 0 assert proposal.symbols_changed == 0 assert proposal.touched_symbols == [] def test_musehub_auth_key_defaults_at_construction() -> None: key = MusehubAuthKey( key_id=_KEY_ID, identity_id=_IDENTITY_ID, public_key_b64="ed25519:AAAA", fingerprint=fake_id("fp"), ) assert key.algorithm == "ed25519" assert key.label == "" assert isinstance(key.created_at, datetime) assert key.last_used_at is None def test_musehub_collaborator_defaults_at_construction() -> None: collab = MusehubCollaborator( id=_COLLAB_ID, repo_id=_REPO_ID, identity_handle="alice", ) assert collab.permission == "write" assert isinstance(collab.invited_at, datetime) assert collab.invited_by_handle is None assert collab.accepted_at is None def test_musehub_branch_defaults_at_construction() -> None: branch = MusehubBranch( branch_id=_BRANCH_ID, repo_id=_REPO_ID, name="main", ) assert branch.head_commit_id is None def test_musehub_label_defaults_at_construction() -> None: from musehub.db.musehub_label_models import MusehubLabel label = MusehubLabel( id=fake_id("label"), repo_id=_REPO_ID, name="bug", color="#d73a4a", ) assert label.description is None assert isinstance(label.created_at, datetime) def test_musehub_commit_defaults_at_construction() -> None: commit = MusehubCommit( commit_id=_COMMIT_ID, branch="main", message="feat: test", author="gabriel", timestamp=datetime.now(tz=timezone.utc), ) assert commit.parent_ids == [] assert commit.snapshot_id is None assert commit.agent_id == "" assert commit.model_id == "" assert commit.toolchain_id == "" assert commit.commit_branch is None assert commit.signature == "" assert commit.signer_public_key == "" assert commit.signer_key_id == "" assert commit.sem_ver_bump == "none" assert commit.breaking_changes == [] assert commit.reviewed_by == [] assert commit.test_runs == 0 assert commit.prompt_hash == "" assert commit.structured_delta is None assert isinstance(commit.created_at, datetime) # --------------------------------------------------------------------------- # Phase 3 — classes that need MappedAsDataclass (mutable callable defaults) # --------------------------------------------------------------------------- _EVENT_ID = fake_id("event") _WEBHOOK_ID = fake_id("webhook") _SESSION_ID = fake_id("session") _COMMENT_ID = fake_id("pcomment") _RELEASE_ID = fake_id("release") _MIST_ID = fake_id("mist") def test_musehub_issue_event_defaults_at_construction() -> None: event = MusehubIssueEvent( event_id=_EVENT_ID, issue_id=_ISSUE_ID, repo_id=_REPO_ID, event_type="opened", ) assert event.actor == "" assert event.payload == {} assert isinstance(event.created_at, datetime) def test_musehub_issue_event_payload_is_isolated() -> None: e1 = MusehubIssueEvent(event_id=_EVENT_ID, issue_id=_ISSUE_ID, repo_id=_REPO_ID, event_type="opened") e2 = MusehubIssueEvent(event_id=_EVENT_ID, issue_id=_ISSUE_ID, repo_id=_REPO_ID, event_type="opened") e1.payload["key"] = "value" assert e2.payload == {}, "payload must not be shared between instances" def test_musehub_webhook_defaults_at_construction() -> None: wh = MusehubWebhook( webhook_id=_WEBHOOK_ID, repo_id=_REPO_ID, url="https://ci.example.com/hook", ) assert wh.events == [] assert wh.secret == "" assert wh.active is True assert isinstance(wh.created_at, datetime) assert isinstance(wh.updated_at, datetime) def test_musehub_webhook_events_is_isolated() -> None: w1 = MusehubWebhook(webhook_id=_WEBHOOK_ID, repo_id=_REPO_ID, url="https://a.example.com") w2 = MusehubWebhook(webhook_id=_WEBHOOK_ID, repo_id=_REPO_ID, url="https://b.example.com") w1.events.append("push") assert w2.events == [], "events must not be shared between instances" def test_musehub_session_defaults_at_construction() -> None: session = MusehubSession( session_id=_SESSION_ID, repo_id=_REPO_ID, started_at=datetime.now(tz=timezone.utc), ) assert session.participants == [] assert session.commits == [] assert session.schema_version == "1" assert session.location == "" assert session.intent == "" assert session.notes == "" assert session.is_active is False assert isinstance(session.created_at, datetime) def test_musehub_session_lists_are_isolated() -> None: now = datetime.now(tz=timezone.utc) s1 = MusehubSession(session_id=_SESSION_ID, repo_id=_REPO_ID, started_at=now) s2 = MusehubSession(session_id=_SESSION_ID, repo_id=_REPO_ID, started_at=now) s1.participants.append("gabriel") assert s2.participants == [], "participants must not be shared between instances" def test_musehub_proposal_comment_defaults_at_construction() -> None: comment = MusehubProposalComment( comment_id=_COMMENT_ID, proposal_id=_PROPOSAL_ID, repo_id=_REPO_ID, author="gabriel", body="looks good", ) assert comment.dimension_ref == {} assert comment.symbol_address is None assert comment.parent_comment_id is None assert isinstance(comment.created_at, datetime) def test_musehub_proposal_comment_dimension_ref_is_isolated() -> None: kwargs = dict(comment_id=_COMMENT_ID, proposal_id=_PROPOSAL_ID, repo_id=_REPO_ID, author="x", body="y") c1 = MusehubProposalComment(**kwargs) c2 = MusehubProposalComment(**kwargs) c1.dimension_ref["dim"] = "harmony" assert c2.dimension_ref == {}, "dimension_ref must not be shared between instances" def test_musehub_release_defaults_at_construction() -> None: release = MusehubRelease( release_id=_RELEASE_ID, repo_id=_REPO_ID, tag="v1.0.0", ) assert release.download_urls == {} assert release.title == "" assert release.body == "" assert release.is_draft is False assert release.channel == "stable" assert release.semver_major == 0 assert isinstance(release.created_at, datetime) assert isinstance(release.updated_at, datetime) def test_musehub_release_download_urls_is_isolated() -> None: r1 = MusehubRelease(release_id=_RELEASE_ID, repo_id=_REPO_ID, tag="v1.0.0") r2 = MusehubRelease(release_id=_RELEASE_ID, repo_id=_REPO_ID, tag="v1.0.0") r1.download_urls["linux"] = "https://example.com/linux.tar.gz" assert r2.download_urls == {}, "download_urls must not be shared between instances" def test_musehub_mist_defaults_at_construction() -> None: mist = MusehubMist( mist_id=_MIST_ID, repo_id=_REPO_ID, owner="gabriel", filename="track.mid", content="", ) assert mist.tags == [] assert mist.symbol_anchors == [] assert mist.artifact_type == "unknown" assert mist.visibility == "public" assert mist.fork_depth == 0 assert mist.view_count == 0 assert isinstance(mist.created_at, datetime) assert isinstance(mist.updated_at, datetime) def test_musehub_mist_lists_are_isolated() -> None: kwargs = dict(mist_id=_MIST_ID, repo_id=_REPO_ID, owner="gabriel", filename="f.mid", content="x") m1 = MusehubMist(**kwargs) m2 = MusehubMist(**kwargs) m1.tags.append("jazz") assert m2.tags == [], "tags must not be shared between instances" def test_musehub_symbol_intel_defaults_at_construction() -> None: intel = MusehubSymbolIntel( repo_id=_REPO_ID, address="src/engine.py::AudioEngine", ) assert intel.blast_top == [] assert intel.weekly == [] assert intel.churn == 0 assert intel.blast == 0 assert intel.gravity == 0.0 def test_musehub_symbol_intel_arrays_are_isolated() -> None: i1 = MusehubSymbolIntel(repo_id=_REPO_ID, address="a.py::Foo") i2 = MusehubSymbolIntel(repo_id=_REPO_ID, address="b.py::Bar") i1.blast_top.append("x.py::Dep") assert i2.blast_top == [], "blast_top must not be shared between instances"