8b84378fe1
Elevate MiroFish/CrowdSight from single-container dev to a SaaS foundation: - Local memory backend (Zep-compatible): memory services/models, local graph builder + updater, AgentActivity seam, import-boundary isolation; Zep stays default, local is opt-in behind MEMORY_BACKEND. Semantic parity not yet proven. - Durable product persistence: projects/simulations/reports schema (migration 0007) + tenant/owner-scoped ProductRepository + dual-write + scoped_project read-first + ArtifactStore abstraction; durable JobQueue + worker.py. - SaaS hardening: durable RateLimiter (wired to login), UsageService (LLM accounting), redacted AuditService, idempotency, CORS allowlist, safe API errors, single-use PasswordResetService + endpoints (covers invite-pending). - Exactly 3 roles (super_admin/admin/user) with tenant authz policy. - Admin UI: GET/POST/PATCH /api/admin/users + GET/PUT /api/admin/settings (super-admin only, encrypted/masked); AdminView.vue + SettingsView.vue with admin/super-admin route guards, th/en i18n. - Production deploy topology: multi-stage Dockerfile (frontend build + gunicorn wsgi + nginx SPA-proxy + supervisord worker), backend/wsgi.py, gunicorn dep. Backend 197 passed; frontend 10 tests + build green. ruff unavailable (gap). No commit of credentials; secrets handled via env/.env.example. Deferred: Zep semantic A/B parity, object storage cutover, mobile QA, EasyPanel container build of deploy topology.
85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
from sqlalchemy import create_engine, event, inspect
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db import Base
|
|
from app.security.policy import Role
|
|
from app.services.identity import IdentityRepository, PasswordService
|
|
|
|
|
|
def make_session():
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
return engine, Session(engine)
|
|
|
|
|
|
def seed_org_and_users(session):
|
|
repo = IdentityRepository(session)
|
|
org_a = repo.create_organization(name="Org A", slug="org-a")
|
|
org_b = repo.create_organization(name="Org B", slug="org-b")
|
|
user_a = repo.create_user(email="user-a@example.com")
|
|
repo.create_membership(user_a.id, org_a.id, Role.USER)
|
|
session.commit()
|
|
return repo, org_a, org_b, user_a
|
|
|
|
|
|
def test_schema_exposes_only_the_three_product_roles():
|
|
assert {role.value for role in Role} == {"super_admin", "admin", "user"}
|
|
|
|
|
|
def test_repository_reads_are_tenant_scoped():
|
|
engine, session = make_session()
|
|
try:
|
|
repo, org_a, org_b, user_a = seed_org_and_users(session)
|
|
assert repo.get_user_for_org(user_a.id, org_a.id) is not None
|
|
assert repo.get_user_for_org(user_a.id, org_b.id) is None
|
|
assert repo.list_users(org_b.id) == []
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
def test_membership_is_unique_per_user_and_organization():
|
|
engine, session = make_session()
|
|
try:
|
|
repo, org_a, _org_b, user_a = seed_org_and_users(session)
|
|
try:
|
|
repo.create_membership(user_a.id, org_a.id, Role.USER)
|
|
session.commit()
|
|
except IntegrityError:
|
|
session.rollback()
|
|
else:
|
|
raise AssertionError("duplicate membership was accepted")
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
def test_repository_does_not_commit_inside_mutations():
|
|
engine, session = make_session()
|
|
commits = []
|
|
|
|
def on_before_commit(*_args):
|
|
commits.append(True)
|
|
|
|
event.listen(Session, "before_commit", on_before_commit)
|
|
try:
|
|
repo = IdentityRepository(session)
|
|
repo.create_organization(name="Org A", slug="org-a")
|
|
repo.create_user(email="user-a@example.com")
|
|
assert commits == []
|
|
session.commit()
|
|
assert len(commits) == 1
|
|
finally:
|
|
event.remove(Session, "before_commit", on_before_commit)
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
def test_password_service_hashes_and_verifies_without_plaintext_storage():
|
|
password = "correct horse battery staple"
|
|
hashed = PasswordService.hash_password(password)
|
|
assert hashed != password
|
|
assert PasswordService.verify_password(hashed, password) is True
|
|
assert PasswordService.verify_password(hashed, "wrong password") is False
|