feat: implement all 4 ORM models (SshKey, Server, Repo, SyncLog)
- Created SshKey model with encrypted private key storage - Created Server model with Gitea configuration and SshKey relationship - Created Repo model with repository mirror info and Server relationship - Created SyncLog model with sync operation logs and Repo relationship - Updated models/__init__.py to export all models - All models use Integer (Unix timestamp) for datetime fields - Proper bidirectional relationships using back_populates - Added comprehensive test suite for all models and relationships Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
panw
0
backend/tests/test_models/__init__.py
Normal file
0
backend/tests/test_models/__init__.py
Normal file
403
backend/tests/test_models/test_orm_models.py
Normal file
403
backend/tests/test_models/test_orm_models.py
Normal file
@@ -0,0 +1,403 @@
|
||||
"""
|
||||
ORM 模型测试.
|
||||
测试所有 4 个模型及其关系.
|
||||
"""
|
||||
import pytest
|
||||
from datetime import datetime
|
||||
from sqlalchemy import inspect
|
||||
from app.models import SshKey, Server, Repo, SyncLog
|
||||
|
||||
|
||||
class TestSshKey:
|
||||
"""测试 SshKey 模型."""
|
||||
|
||||
def test_create_ssh_key(self, db_session):
|
||||
"""测试创建 SSH 密钥."""
|
||||
ssh_key = SshKey(
|
||||
name="test-key",
|
||||
private_key="encrypted-private-key-content",
|
||||
fingerprint="SHA256:abc123",
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(ssh_key)
|
||||
db_session.commit()
|
||||
db_session.refresh(ssh_key)
|
||||
|
||||
assert ssh_key.id is not None
|
||||
assert ssh_key.name == "test-key"
|
||||
assert ssh_key.private_key == "encrypted-private-key-content"
|
||||
assert ssh_key.fingerprint == "SHA256:abc123"
|
||||
assert isinstance(ssh_key.created_at, int)
|
||||
|
||||
def test_ssh_key_table_structure(self, db_engine):
|
||||
"""测试 ssh_keys 表结构."""
|
||||
inspector = inspect(db_engine)
|
||||
columns = [col['name'] for col in inspector.get_columns('ssh_keys')]
|
||||
|
||||
assert 'id' in columns
|
||||
assert 'name' in columns
|
||||
assert 'private_key' in columns
|
||||
assert 'fingerprint' in columns
|
||||
assert 'created_at' in columns
|
||||
|
||||
|
||||
class TestServer:
|
||||
"""测试 Server 模型."""
|
||||
|
||||
def test_create_server(self, db_session):
|
||||
"""测试创建服务器."""
|
||||
server = Server(
|
||||
name="test-server",
|
||||
url="https://gitea.example.com",
|
||||
api_token="encrypted-api-token",
|
||||
ssh_key_id=1,
|
||||
sync_enabled=True,
|
||||
schedule_cron="0 */2 * * *",
|
||||
local_path="/data/repos/test-server",
|
||||
status="connected",
|
||||
created_at=int(datetime.utcnow().timestamp()),
|
||||
updated_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(server)
|
||||
db_session.commit()
|
||||
db_session.refresh(server)
|
||||
|
||||
assert server.id is not None
|
||||
assert server.name == "test-server"
|
||||
assert server.url == "https://gitea.example.com"
|
||||
assert server.api_token == "encrypted-api-token"
|
||||
assert server.ssh_key_id == 1
|
||||
assert server.sync_enabled is True
|
||||
assert server.schedule_cron == "0 */2 * * *"
|
||||
assert server.local_path == "/data/repos/test-server"
|
||||
assert server.status == "connected"
|
||||
assert isinstance(server.created_at, int)
|
||||
assert isinstance(server.updated_at, int)
|
||||
|
||||
def test_server_table_structure(self, db_engine):
|
||||
"""测试 servers 表结构."""
|
||||
inspector = inspect(db_engine)
|
||||
columns = [col['name'] for col in inspector.get_columns('servers')]
|
||||
|
||||
assert 'id' in columns
|
||||
assert 'name' in columns
|
||||
assert 'url' in columns
|
||||
assert 'api_token' in columns
|
||||
assert 'ssh_key_id' in columns
|
||||
assert 'sync_enabled' in columns
|
||||
assert 'schedule_cron' in columns
|
||||
assert 'local_path' in columns
|
||||
assert 'status' in columns
|
||||
assert 'created_at' in columns
|
||||
assert 'updated_at' in columns
|
||||
|
||||
def test_server_ssh_key_relationship(self, db_session):
|
||||
"""测试 Server 和 SshKey 的关系."""
|
||||
# 先创建 SSH 密钥
|
||||
ssh_key = SshKey(
|
||||
name="test-key",
|
||||
private_key="encrypted-key",
|
||||
fingerprint="SHA256:abc123",
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(ssh_key)
|
||||
db_session.commit()
|
||||
|
||||
# 创建服务器并关联 SSH 密钥
|
||||
server = Server(
|
||||
name="test-server",
|
||||
url="https://gitea.example.com",
|
||||
api_token="encrypted-token",
|
||||
ssh_key_id=ssh_key.id,
|
||||
sync_enabled=False,
|
||||
local_path="/data/repos",
|
||||
status="untested",
|
||||
created_at=int(datetime.utcnow().timestamp()),
|
||||
updated_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(server)
|
||||
db_session.commit()
|
||||
db_session.refresh(server)
|
||||
|
||||
# 测试关系
|
||||
assert server.ssh_key_id == ssh_key.id
|
||||
assert server.ssh_key.id == ssh_key.id
|
||||
assert server.ssh_key.name == "test-key"
|
||||
|
||||
|
||||
class TestRepo:
|
||||
"""测试 Repo 模型."""
|
||||
|
||||
def test_create_repo(self, db_session):
|
||||
"""测试创建仓库."""
|
||||
repo = Repo(
|
||||
server_id=1,
|
||||
name="test-repo",
|
||||
full_name="owner/test-repo",
|
||||
clone_url="git@gitea.example.com:owner/test-repo.git",
|
||||
local_path="/data/repos/test-server/owner/test-repo",
|
||||
last_sync_at=int(datetime.utcnow().timestamp()),
|
||||
status="synced",
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(repo)
|
||||
db_session.commit()
|
||||
db_session.refresh(repo)
|
||||
|
||||
assert repo.id is not None
|
||||
assert repo.server_id == 1
|
||||
assert repo.name == "test-repo"
|
||||
assert repo.full_name == "owner/test-repo"
|
||||
assert repo.clone_url == "git@gitea.example.com:owner/test-repo.git"
|
||||
assert repo.local_path == "/data/repos/test-server/owner/test-repo"
|
||||
assert isinstance(repo.last_sync_at, int)
|
||||
assert repo.status == "synced"
|
||||
assert isinstance(repo.created_at, int)
|
||||
|
||||
def test_repo_table_structure(self, db_engine):
|
||||
"""测试 repos 表结构."""
|
||||
inspector = inspect(db_engine)
|
||||
columns = [col['name'] for col in inspector.get_columns('repos')]
|
||||
|
||||
assert 'id' in columns
|
||||
assert 'server_id' in columns
|
||||
assert 'name' in columns
|
||||
assert 'full_name' in columns
|
||||
assert 'clone_url' in columns
|
||||
assert 'local_path' in columns
|
||||
assert 'last_sync_at' in columns
|
||||
assert 'status' in columns
|
||||
assert 'created_at' in columns
|
||||
|
||||
def test_repo_server_relationship(self, db_session):
|
||||
"""测试 Repo 和 Server 的关系."""
|
||||
# 先创建 SSH 密钥
|
||||
ssh_key = SshKey(
|
||||
name="test-key",
|
||||
private_key="encrypted-key",
|
||||
fingerprint="SHA256:abc123",
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(ssh_key)
|
||||
db_session.commit()
|
||||
|
||||
# 创建服务器
|
||||
server = Server(
|
||||
name="test-server",
|
||||
url="https://gitea.example.com",
|
||||
api_token="encrypted-token",
|
||||
ssh_key_id=ssh_key.id,
|
||||
sync_enabled=False,
|
||||
local_path="/data/repos",
|
||||
status="untested",
|
||||
created_at=int(datetime.utcnow().timestamp()),
|
||||
updated_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(server)
|
||||
db_session.commit()
|
||||
|
||||
# 创建仓库并关联服务器
|
||||
repo = Repo(
|
||||
server_id=server.id,
|
||||
name="test-repo",
|
||||
full_name="owner/test-repo",
|
||||
clone_url="git@gitea.example.com:owner/test-repo.git",
|
||||
local_path="/data/repos/test-server/owner/test-repo",
|
||||
last_sync_at=int(datetime.utcnow().timestamp()),
|
||||
status="synced",
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(repo)
|
||||
db_session.commit()
|
||||
db_session.refresh(repo)
|
||||
|
||||
# 测试关系
|
||||
assert repo.server_id == server.id
|
||||
assert repo.server.id == server.id
|
||||
assert repo.server.name == "test-server"
|
||||
assert len(server.repos) == 1
|
||||
assert server.repos[0].name == "test-repo"
|
||||
|
||||
|
||||
class TestSyncLog:
|
||||
"""测试 SyncLog 模型."""
|
||||
|
||||
def test_create_sync_log(self, db_session):
|
||||
"""测试创建同步日志."""
|
||||
sync_log = SyncLog(
|
||||
repo_id=1,
|
||||
status="synced",
|
||||
started_at=int(datetime.utcnow().timestamp()),
|
||||
finished_at=int(datetime.utcnow().timestamp()),
|
||||
commits_count=5,
|
||||
error_msg=None,
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(sync_log)
|
||||
db_session.commit()
|
||||
db_session.refresh(sync_log)
|
||||
|
||||
assert sync_log.id is not None
|
||||
assert sync_log.repo_id == 1
|
||||
assert sync_log.status == "synced"
|
||||
assert isinstance(sync_log.started_at, int)
|
||||
assert isinstance(sync_log.finished_at, int)
|
||||
assert sync_log.commits_count == 5
|
||||
assert sync_log.error_msg is None
|
||||
assert isinstance(sync_log.created_at, int)
|
||||
|
||||
def test_sync_log_table_structure(self, db_engine):
|
||||
"""测试 sync_logs 表结构."""
|
||||
inspector = inspect(db_engine)
|
||||
columns = [col['name'] for col in inspector.get_columns('sync_logs')]
|
||||
|
||||
assert 'id' in columns
|
||||
assert 'repo_id' in columns
|
||||
assert 'status' in columns
|
||||
assert 'started_at' in columns
|
||||
assert 'finished_at' in columns
|
||||
assert 'commits_count' in columns
|
||||
assert 'error_msg' in columns
|
||||
assert 'created_at' in columns
|
||||
|
||||
def test_sync_log_repo_relationship(self, db_session):
|
||||
"""测试 SyncLog 和 Repo 的关系."""
|
||||
# 先创建 SSH 密钥
|
||||
ssh_key = SshKey(
|
||||
name="test-key",
|
||||
private_key="encrypted-key",
|
||||
fingerprint="SHA256:abc123",
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(ssh_key)
|
||||
db_session.commit()
|
||||
|
||||
# 创建服务器
|
||||
server = Server(
|
||||
name="test-server",
|
||||
url="https://gitea.example.com",
|
||||
api_token="encrypted-token",
|
||||
ssh_key_id=ssh_key.id,
|
||||
sync_enabled=False,
|
||||
local_path="/data/repos",
|
||||
status="untested",
|
||||
created_at=int(datetime.utcnow().timestamp()),
|
||||
updated_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(server)
|
||||
db_session.commit()
|
||||
|
||||
# 创建仓库
|
||||
repo = Repo(
|
||||
server_id=server.id,
|
||||
name="test-repo",
|
||||
full_name="owner/test-repo",
|
||||
clone_url="git@gitea.example.com:owner/test-repo.git",
|
||||
local_path="/data/repos/test-server/owner/test-repo",
|
||||
last_sync_at=int(datetime.utcnow().timestamp()),
|
||||
status="synced",
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(repo)
|
||||
db_session.commit()
|
||||
|
||||
# 创建同步日志
|
||||
sync_log = SyncLog(
|
||||
repo_id=repo.id,
|
||||
status="synced",
|
||||
started_at=int(datetime.utcnow().timestamp()),
|
||||
finished_at=int(datetime.utcnow().timestamp()),
|
||||
commits_count=10,
|
||||
error_msg=None,
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(sync_log)
|
||||
db_session.commit()
|
||||
db_session.refresh(sync_log)
|
||||
|
||||
# 测试关系
|
||||
assert sync_log.repo_id == repo.id
|
||||
assert sync_log.repo.id == repo.id
|
||||
assert sync_log.repo.name == "test-repo"
|
||||
assert len(repo.sync_logs) == 1
|
||||
assert repo.sync_logs[0].commits_count == 10
|
||||
|
||||
|
||||
class TestModelRelationships:
|
||||
"""测试完整的模型关系链."""
|
||||
|
||||
def test_full_relationship_chain(self, db_session):
|
||||
"""测试 SshKey -> Server -> Repo -> SyncLog 完整关系链."""
|
||||
# 创建 SSH 密钥
|
||||
ssh_key = SshKey(
|
||||
name="deploy-key",
|
||||
private_key="encrypted-private-key",
|
||||
fingerprint="SHA256:xyz789",
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(ssh_key)
|
||||
db_session.commit()
|
||||
|
||||
# 创建服务器
|
||||
server = Server(
|
||||
name="gitea-server",
|
||||
url="https://gitea.example.com",
|
||||
api_token="encrypted-api-token",
|
||||
ssh_key_id=ssh_key.id,
|
||||
sync_enabled=True,
|
||||
schedule_cron="0 */2 * * *",
|
||||
local_path="/data/repos/gitea-server",
|
||||
status="connected",
|
||||
created_at=int(datetime.utcnow().timestamp()),
|
||||
updated_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(server)
|
||||
db_session.commit()
|
||||
|
||||
# 创建仓库
|
||||
repo = Repo(
|
||||
server_id=server.id,
|
||||
name="my-project",
|
||||
full_name="john/my-project",
|
||||
clone_url="git@gitea.example.com:john/my-project.git",
|
||||
local_path="/data/repos/gitea-server/john/my-project",
|
||||
last_sync_at=int(datetime.utcnow().timestamp()),
|
||||
status="synced",
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(repo)
|
||||
db_session.commit()
|
||||
|
||||
# 创建同步日志
|
||||
sync_log = SyncLog(
|
||||
repo_id=repo.id,
|
||||
status="synced",
|
||||
started_at=int(datetime.utcnow().timestamp()),
|
||||
finished_at=int(datetime.utcnow().timestamp()),
|
||||
commits_count=15,
|
||||
error_msg=None,
|
||||
created_at=int(datetime.utcnow().timestamp())
|
||||
)
|
||||
db_session.add(sync_log)
|
||||
db_session.commit()
|
||||
|
||||
# 验证关系链
|
||||
db_session.refresh(ssh_key)
|
||||
db_session.refresh(server)
|
||||
db_session.refresh(repo)
|
||||
db_session.refresh(sync_log)
|
||||
|
||||
# SshKey -> Server
|
||||
assert len(ssh_key.servers) == 1
|
||||
assert ssh_key.servers[0].name == "gitea-server"
|
||||
|
||||
# Server -> Repo
|
||||
assert len(server.repos) == 1
|
||||
assert server.repos[0].name == "my-project"
|
||||
|
||||
# Repo -> SyncLog
|
||||
assert len(repo.sync_logs) == 1
|
||||
assert repo.sync_logs[0].commits_count == 15
|
||||
|
||||
# 反向关系
|
||||
assert sync_log.repo.server.ssh_key.name == "deploy-key"
|
||||
Reference in New Issue
Block a user