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.
71 lines
2.6 KiB
Python
71 lines
2.6 KiB
Python
"""TDD gate: tenant-scoped artifact store abstraction.
|
|
|
|
The legacy project/simulation/report managers write to raw os.path.join under a
|
|
shared upload root. This gate proves a scoped ArtifactStore resolves paths
|
|
within the tenant's own directory, rejects traversal/absolute components, and
|
|
exposes a store/read/delete interface that can later be backed by object
|
|
storage without changing callers.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from app.services.artifact_store import ArtifactStore
|
|
|
|
|
|
def test_artifact_store_scopes_path_to_root(monkeypatch, tmp_path):
|
|
store = ArtifactStore(str(tmp_path))
|
|
project_dir = store.path_for("org-a", "project-1")
|
|
assert str(tmp_path) in project_dir
|
|
assert "org-a" in project_dir
|
|
assert "project-1" in project_dir
|
|
# Tenant-scoped directory lands under the configured root.
|
|
assert project_dir.startswith(str(tmp_path))
|
|
|
|
|
|
def test_artifact_store_isolates_tenants(monkeypatch, tmp_path):
|
|
store = ArtifactStore(str(tmp_path))
|
|
path_a = store.path_for("org-a", "proj-x", "state.json")
|
|
path_b = store.path_for("org-b", "proj-x", "state.json")
|
|
assert path_a != path_b
|
|
assert "/org-a/" in path_a.replace("\\", "/")
|
|
assert "/org-b/" in path_b.replace("\\", "/")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"bad_segment",
|
|
["../evil", "a/../../etc", "/absolute", "..", ".../..", "a//..", "\\evil", "a/..\\b"],
|
|
)
|
|
def test_artifact_store_rejects_traversal(bad_segment, tmp_path):
|
|
store = ArtifactStore(str(tmp_path))
|
|
with pytest.raises(ValueError):
|
|
store.path_for("org-a", bad_segment)
|
|
|
|
|
|
def test_artifact_store_roundtrip_bytes(monkeypatch, tmp_path):
|
|
store = ArtifactStore(str(tmp_path))
|
|
target = store.path_for("org-a", "report-1", "report.md")
|
|
store.ensure_parent(target)
|
|
store.store_bytes(target, b"# title\nbody")
|
|
assert store.read_bytes(target) == b"# title\nbody"
|
|
assert store.exists(target)
|
|
store.delete(target)
|
|
assert not store.exists(target)
|
|
|
|
|
|
def test_default_artifact_store_uses_configured_upload_root(monkeypatch, tmp_path):
|
|
from app.config import Config
|
|
from app.services.artifact_store import default_artifact_store
|
|
|
|
original = Config.UPLOAD_FOLDER
|
|
monkeypatch.setattr(Config, "UPLOAD_FOLDER", str(tmp_path / "uploads"))
|
|
try:
|
|
store = default_artifact_store()
|
|
project_dir = store.path_for("org-a", "project-1")
|
|
assert str(Config.UPLOAD_FOLDER) in project_dir
|
|
# Tenant-scoped and confinement still apply.
|
|
assert ".." not in project_dir
|
|
with pytest.raises(ValueError):
|
|
store.path_for("org-a", "..")
|
|
finally:
|
|
Config.UPLOAD_FOLDER = original
|