455 lines
16 KiB
Python
455 lines
16 KiB
Python
"""
|
|
Test Recovery Management Functionality Preservation
|
|
|
|
This test suite verifies that all existing API endpoints, navigation, and component
|
|
interfaces continue to work correctly after terminology changes.
|
|
|
|
Requirements tested:
|
|
- 4.1: Component interfaces remain unchanged
|
|
- 4.2: API endpoints continue to work
|
|
- 4.3: Navigation and routing functionality is preserved
|
|
- 4.4: Service method signatures stay the same
|
|
- 4.5: Database schemas remain unchanged
|
|
"""
|
|
|
|
import pytest
|
|
import requests
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime
|
|
|
|
from main import app
|
|
from database import get_db, engine
|
|
from models.user import User, UserRole
|
|
from models.project import Project
|
|
from models.shot import Shot
|
|
from models.asset import Asset, AssetCategory
|
|
from models.task import Task, TaskStatus
|
|
from utils.auth import create_access_token
|
|
from services.recovery_service import RecoveryService
|
|
|
|
client = TestClient(app)
|
|
|
|
# Test data setup
|
|
@pytest.fixture
|
|
def db_session():
|
|
"""Create a test database session"""
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from models import Base
|
|
|
|
# Use in-memory SQLite for testing
|
|
test_engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(bind=test_engine)
|
|
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=test_engine)
|
|
session = TestingSessionLocal()
|
|
|
|
yield session
|
|
|
|
session.close()
|
|
|
|
@pytest.fixture
|
|
def admin_user(db_session):
|
|
"""Create an admin user for testing"""
|
|
admin = User(
|
|
username="admin_test",
|
|
email="admin@test.com",
|
|
full_name="Admin Test User",
|
|
role=UserRole.admin,
|
|
is_admin=True,
|
|
is_approved=True,
|
|
hashed_password="fake_hash"
|
|
)
|
|
db_session.add(admin)
|
|
db_session.commit()
|
|
db_session.refresh(admin)
|
|
return admin
|
|
|
|
@pytest.fixture
|
|
def test_project(db_session):
|
|
"""Create a test project"""
|
|
project = Project(
|
|
name="Test Project",
|
|
description="Test project for recovery testing",
|
|
status="active",
|
|
created_by=1
|
|
)
|
|
db_session.add(project)
|
|
db_session.commit()
|
|
db_session.refresh(project)
|
|
return project
|
|
|
|
@pytest.fixture
|
|
def deleted_shot(db_session, test_project, admin_user):
|
|
"""Create a soft-deleted shot for testing"""
|
|
shot = Shot(
|
|
name="TEST_001",
|
|
episode_name="Episode 1",
|
|
project_id=test_project.id,
|
|
created_by=admin_user.id,
|
|
is_deleted=True,
|
|
deleted_at=datetime.utcnow(),
|
|
deleted_by=admin_user.id
|
|
)
|
|
db_session.add(shot)
|
|
db_session.commit()
|
|
db_session.refresh(shot)
|
|
return shot
|
|
|
|
@pytest.fixture
|
|
def deleted_asset(db_session, test_project, admin_user):
|
|
"""Create a soft-deleted asset for testing"""
|
|
asset = Asset(
|
|
name="TestCharacter",
|
|
category=AssetCategory.character,
|
|
project_id=test_project.id,
|
|
created_by=admin_user.id,
|
|
is_deleted=True,
|
|
deleted_at=datetime.utcnow(),
|
|
deleted_by=admin_user.id
|
|
)
|
|
db_session.add(asset)
|
|
db_session.commit()
|
|
db_session.refresh(asset)
|
|
return asset
|
|
|
|
@pytest.fixture
|
|
def admin_token(admin_user):
|
|
"""Create an admin access token"""
|
|
return create_access_token(data={"user_id": admin_user.id})
|
|
|
|
def override_get_db(db_session):
|
|
"""Override the database dependency for testing"""
|
|
def _get_db():
|
|
try:
|
|
yield db_session
|
|
finally:
|
|
pass
|
|
return _get_db
|
|
|
|
class TestAPIEndpointPreservation:
|
|
"""Test that all existing API endpoints continue to work"""
|
|
|
|
def test_get_deleted_shots_endpoint(self, db_session, admin_token, deleted_shot):
|
|
"""Test that /admin/deleted-shots/ endpoint works correctly"""
|
|
app.dependency_overrides[get_db] = override_get_db(db_session)
|
|
|
|
response = client.get(
|
|
"/api/admin/deleted-shots/",
|
|
headers={"Authorization": f"Bearer {admin_token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
if data: # If there are deleted shots
|
|
shot_data = data[0]
|
|
required_fields = [
|
|
"id", "name", "episode_name", "project_id", "project_name",
|
|
"deleted_at", "deleted_by", "deleted_by_name",
|
|
"task_count", "submission_count", "attachment_count",
|
|
"note_count", "review_count"
|
|
]
|
|
for field in required_fields:
|
|
assert field in shot_data, f"Missing field: {field}"
|
|
|
|
def test_get_deleted_assets_endpoint(self, db_session, admin_token, deleted_asset):
|
|
"""Test that /admin/deleted-assets/ endpoint works correctly"""
|
|
app.dependency_overrides[get_db] = override_get_db(db_session)
|
|
|
|
response = client.get(
|
|
"/api/admin/deleted-assets/",
|
|
headers={"Authorization": f"Bearer {admin_token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
if data: # If there are deleted assets
|
|
asset_data = data[0]
|
|
required_fields = [
|
|
"id", "name", "category", "project_id", "project_name",
|
|
"deleted_at", "deleted_by", "deleted_by_name",
|
|
"task_count", "submission_count", "attachment_count",
|
|
"note_count", "review_count"
|
|
]
|
|
for field in required_fields:
|
|
assert field in asset_data, f"Missing field: {field}"
|
|
|
|
def test_shot_recovery_preview_endpoint(self, db_session, admin_token, deleted_shot):
|
|
"""Test that shot recovery preview endpoint works correctly"""
|
|
app.dependency_overrides[get_db] = override_get_db(db_session)
|
|
|
|
response = client.get(
|
|
f"/api/admin/shots/{deleted_shot.id}/recovery-preview",
|
|
headers={"Authorization": f"Bearer {admin_token}"}
|
|
)
|
|
|
|
# Should return 200 or 404 (if shot not found in recovery data)
|
|
assert response.status_code in [200, 404]
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
required_fields = [
|
|
"shot_id", "name", "episode_name", "project_name",
|
|
"task_count", "submission_count", "attachment_count",
|
|
"note_count", "review_count", "deleted_at", "deleted_by",
|
|
"deleted_by_name", "files_preserved", "file_count"
|
|
]
|
|
for field in required_fields:
|
|
assert field in data, f"Missing field: {field}"
|
|
|
|
def test_asset_recovery_preview_endpoint(self, db_session, admin_token, deleted_asset):
|
|
"""Test that asset recovery preview endpoint works correctly"""
|
|
app.dependency_overrides[get_db] = override_get_db(db_session)
|
|
|
|
response = client.get(
|
|
f"/api/admin/assets/{deleted_asset.id}/recovery-preview",
|
|
headers={"Authorization": f"Bearer {admin_token}"}
|
|
)
|
|
|
|
# Should return 200 or 404 (if asset not found in recovery data)
|
|
assert response.status_code in [200, 404]
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
required_fields = [
|
|
"asset_id", "name", "project_name",
|
|
"task_count", "submission_count", "attachment_count",
|
|
"note_count", "review_count", "deleted_at", "deleted_by",
|
|
"deleted_by_name", "files_preserved", "file_count"
|
|
]
|
|
for field in required_fields:
|
|
assert field in data, f"Missing field: {field}"
|
|
|
|
def test_recovery_stats_endpoint(self, db_session, admin_token):
|
|
"""Test that recovery stats endpoint works correctly"""
|
|
app.dependency_overrides[get_db] = override_get_db(db_session)
|
|
|
|
response = client.get(
|
|
"/api/admin/recovery-stats/",
|
|
headers={"Authorization": f"Bearer {admin_token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
required_fields = [
|
|
"deleted_shots_count", "deleted_assets_count",
|
|
"total_deleted_tasks", "total_deleted_files"
|
|
]
|
|
for field in required_fields:
|
|
assert field in data, f"Missing field: {field}"
|
|
|
|
def test_bulk_recovery_endpoints_structure(self, db_session, admin_token):
|
|
"""Test that bulk recovery endpoints have correct structure"""
|
|
app.dependency_overrides[get_db] = override_get_db(db_session)
|
|
|
|
# Test bulk shot recovery with empty list (should return 400)
|
|
response = client.post(
|
|
"/api/admin/shots/bulk-recover",
|
|
json={"shot_ids": []},
|
|
headers={"Authorization": f"Bearer {admin_token}"}
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
# Test bulk asset recovery with empty list (should return 400)
|
|
response = client.post(
|
|
"/api/admin/assets/bulk-recover",
|
|
json={"asset_ids": []},
|
|
headers={"Authorization": f"Bearer {admin_token}"}
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_endpoint_authentication_required(self):
|
|
"""Test that all endpoints require authentication"""
|
|
endpoints = [
|
|
"/api/admin/deleted-shots/",
|
|
"/api/admin/deleted-assets/",
|
|
"/api/admin/recovery-stats/"
|
|
]
|
|
|
|
for endpoint in endpoints:
|
|
response = client.get(endpoint)
|
|
assert response.status_code == 401, f"Endpoint {endpoint} should require authentication"
|
|
|
|
|
|
class TestServiceInterfacePreservation:
|
|
"""Test that service method signatures remain unchanged"""
|
|
|
|
def test_recovery_service_interface(self, db_session):
|
|
"""Test that RecoveryService maintains its interface"""
|
|
service = RecoveryService()
|
|
|
|
# Test that all expected methods exist
|
|
expected_methods = [
|
|
'get_deleted_shots',
|
|
'get_deleted_assets',
|
|
'preview_shot_recovery',
|
|
'preview_asset_recovery',
|
|
'recover_shot',
|
|
'recover_asset',
|
|
'bulk_recover_shots',
|
|
'bulk_recover_assets',
|
|
'get_recovery_stats'
|
|
]
|
|
|
|
for method_name in expected_methods:
|
|
assert hasattr(service, method_name), f"Missing method: {method_name}"
|
|
method = getattr(service, method_name)
|
|
assert callable(method), f"Method {method_name} is not callable"
|
|
|
|
def test_recovery_service_method_signatures(self, db_session):
|
|
"""Test that method signatures haven't changed"""
|
|
service = RecoveryService()
|
|
|
|
# Test get_deleted_shots signature
|
|
try:
|
|
# Should accept project_id and db parameters
|
|
service.get_deleted_shots(None, db_session)
|
|
except Exception as e:
|
|
# Method should exist and accept these parameters
|
|
# Any other error is acceptable for this interface test
|
|
assert "missing" not in str(e).lower()
|
|
|
|
# Test get_deleted_assets signature
|
|
try:
|
|
service.get_deleted_assets(None, db_session)
|
|
except Exception as e:
|
|
assert "missing" not in str(e).lower()
|
|
|
|
|
|
class TestComponentInterfacePreservation:
|
|
"""Test that component interfaces remain unchanged"""
|
|
|
|
def test_recovery_service_types_preserved(self):
|
|
"""Test that TypeScript interfaces are preserved"""
|
|
# This test verifies that the service file structure is maintained
|
|
import os
|
|
|
|
service_file = "frontend/src/services/recovery.ts"
|
|
assert os.path.exists(service_file), "Recovery service file should exist"
|
|
|
|
with open(service_file, 'r') as f:
|
|
content = f.read()
|
|
|
|
# Check that key interfaces are still defined
|
|
required_interfaces = [
|
|
"DeletedShot",
|
|
"DeletedAsset",
|
|
"RecoveryInfo",
|
|
"RecoveryResult",
|
|
"BulkRecoveryResult"
|
|
]
|
|
|
|
for interface in required_interfaces:
|
|
assert f"interface {interface}" in content, f"Missing interface: {interface}"
|
|
|
|
# Check that key methods are still defined
|
|
required_methods = [
|
|
"getDeletedShots",
|
|
"getDeletedAssets",
|
|
"previewShotRecovery",
|
|
"previewAssetRecovery",
|
|
"recoverShot",
|
|
"recoverAsset",
|
|
"bulkRecoverShots",
|
|
"bulkRecoverAssets"
|
|
]
|
|
|
|
for method in required_methods:
|
|
assert method in content, f"Missing method: {method}"
|
|
|
|
|
|
class TestNavigationAndRoutingPreservation:
|
|
"""Test that navigation and routing functionality is preserved"""
|
|
|
|
def test_router_configuration_preserved(self):
|
|
"""Test that router configuration maintains existing routes"""
|
|
import os
|
|
|
|
router_file = "frontend/src/router/index.ts"
|
|
assert os.path.exists(router_file), "Router file should exist"
|
|
|
|
with open(router_file, 'r') as f:
|
|
content = f.read()
|
|
|
|
# Check that the recovery management route exists
|
|
assert "/admin/deleted-items" in content, "Recovery management route should exist"
|
|
assert "RecoveryManagement" in content, "RecoveryManagement route name should exist"
|
|
assert "DeletedItemsManagementView" in content, "Component import should exist"
|
|
|
|
def test_sidebar_navigation_preserved(self):
|
|
"""Test that sidebar navigation structure is preserved"""
|
|
import os
|
|
|
|
sidebar_file = "frontend/src/components/layout/AppSidebar.vue"
|
|
assert os.path.exists(sidebar_file), "Sidebar file should exist"
|
|
|
|
with open(sidebar_file, 'r') as f:
|
|
content = f.read()
|
|
|
|
# Check that admin navigation section exists
|
|
assert "adminItems" in content, "Admin items should be defined"
|
|
assert "/admin/deleted-items" in content, "Recovery management URL should exist"
|
|
assert "RotateCcw" in content, "Recovery icon should be imported"
|
|
|
|
def test_component_file_structure_preserved(self):
|
|
"""Test that component file structure is maintained"""
|
|
import os
|
|
|
|
# Check that the main component file exists
|
|
component_file = "frontend/src/views/admin/DeletedItemsManagementView.vue"
|
|
assert os.path.exists(component_file), "Main recovery component should exist"
|
|
|
|
# Check that the component has the expected structure
|
|
with open(component_file, 'r') as f:
|
|
content = f.read()
|
|
|
|
# Verify key component elements are present
|
|
assert "<template>" in content, "Component should have template section"
|
|
assert "<script setup" in content, "Component should use composition API"
|
|
assert "recoveryService" in content, "Component should use recovery service"
|
|
|
|
|
|
class TestDatabaseSchemaPreservation:
|
|
"""Test that database schemas remain unchanged"""
|
|
|
|
def test_shot_model_schema_preserved(self, db_session):
|
|
"""Test that Shot model maintains required fields"""
|
|
from models.shot import Shot
|
|
from sqlalchemy import inspect
|
|
|
|
inspector = inspect(Shot)
|
|
column_names = [column.name for column in inspector.columns]
|
|
|
|
required_fields = [
|
|
'id', 'name', 'episode_name', 'project_id',
|
|
'is_deleted', 'deleted_at', 'deleted_by'
|
|
]
|
|
|
|
for field in required_fields:
|
|
assert field in column_names, f"Shot model missing field: {field}"
|
|
|
|
def test_asset_model_schema_preserved(self, db_session):
|
|
"""Test that Asset model maintains required fields"""
|
|
from models.asset import Asset
|
|
from sqlalchemy import inspect
|
|
|
|
inspector = inspect(Asset)
|
|
column_names = [column.name for column in inspector.columns]
|
|
|
|
required_fields = [
|
|
'id', 'name', 'category', 'project_id',
|
|
'is_deleted', 'deleted_at', 'deleted_by'
|
|
]
|
|
|
|
for field in required_fields:
|
|
assert field in column_names, f"Asset model missing field: {field}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Run the tests
|
|
pytest.main([__file__, "-v"]) |