From f425a49773928084d844d0aa19602b05f004690f Mon Sep 17 00:00:00 2001
From: panw <panfudan0209@qq.com>
Date: Mon, 30 Mar 2026 15:37:36 +0800
Subject: [PATCH] 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>
---
 backend/app/models/__init__.py               |   6 +-
 backend/app/models/repo.py                   |  33 ++
 backend/app/models/server.py                 |  35 ++
 backend/app/models/ssh_key.py                |  28 ++
 backend/app/models/sync_log.py               |  31 ++
 backend/tests/test_models/__init__.py        |   0
 backend/tests/test_models/test_orm_models.py | 403 +++++++++++++++++++
 7 files changed, 535 insertions(+), 1 deletion(-)
 create mode 100644 backend/app/models/repo.py
 create mode 100644 backend/app/models/server.py
 create mode 100644 backend/app/models/ssh_key.py
 create mode 100644 backend/app/models/sync_log.py
 create mode 100644 backend/tests/test_models/__init__.py
 create mode 100644 backend/tests/test_models/test_orm_models.py

diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py
index 57056e3..ae034f1 100644
--- a/backend/app/models/__init__.py
+++ b/backend/app/models/__init__.py
@@ -1,3 +1,7 @@
 from app.database import Base
+from app.models.ssh_key import SshKey
+from app.models.server import Server
+from app.models.repo import Repo
+from app.models.sync_log import SyncLog
 
-__all__ = ['Base']
+__all__ = ['Base', 'SshKey', 'Server', 'Repo', 'SyncLog']
diff --git a/backend/app/models/repo.py b/backend/app/models/repo.py
new file mode 100644
index 0000000..deb9779
--- /dev/null
+++ b/backend/app/models/repo.py
@@ -0,0 +1,33 @@
+"""
+仓库 ORM 模型.
+"""
+from sqlalchemy import String, Integer, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from app.database import Base
+from typing import Optional, List
+
+
+class Repo(Base):
+    """
+    Git 仓库镜像模型.
+
+    存储从 Gitea 服务器同步的仓库信息.
+    """
+    __tablename__ = "repos"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    server_id: Mapped[int] = mapped_column(Integer, ForeignKey("servers.id"), nullable=False)
+    name: Mapped[str] = mapped_column(String(200), nullable=False)
+    full_name: Mapped[str] = mapped_column(String(300), nullable=False)
+    clone_url: Mapped[str] = mapped_column(String(500), nullable=False)
+    local_path: Mapped[str] = mapped_column(String(500), nullable=False)
+    last_sync_at: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
+    created_at: Mapped[int] = mapped_column(Integer, nullable=False)
+
+    # 关系
+    server: Mapped["Server"] = relationship("Server", back_populates="repos")
+    sync_logs: Mapped[List["SyncLog"]] = relationship("SyncLog", back_populates="repo", cascade="all, delete-orphan")
+
+    def __repr__(self) -> str:
+        return f"<Repo(id={self.id}, name='{self.name}', full_name='{self.full_name}', status='{self.status}')>"
diff --git a/backend/app/models/server.py b/backend/app/models/server.py
new file mode 100644
index 0000000..26bf561
--- /dev/null
+++ b/backend/app/models/server.py
@@ -0,0 +1,35 @@
+"""
+服务器 ORM 模型.
+"""
+from sqlalchemy import String, Integer, Text, Boolean, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from app.database import Base
+from typing import Optional, List
+
+
+class Server(Base):
+    """
+    Gitea 服务器模型.
+
+    存储 Gitea 服务器配置和连接信息.
+    """
+    __tablename__ = "servers"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(100), nullable=False, unique=True)
+    url: Mapped[str] = mapped_column(String(500), nullable=False)
+    api_token: Mapped[str] = mapped_column(Text, nullable=False)
+    ssh_key_id: Mapped[int] = mapped_column(Integer, ForeignKey("ssh_keys.id"), nullable=False)
+    sync_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    schedule_cron: Mapped[Optional[str]] = mapped_column(String(50), nullable=True)
+    local_path: Mapped[str] = mapped_column(String(500), nullable=False)
+    status: Mapped[str] = mapped_column(String(20), nullable=False, default="untested")
+    created_at: Mapped[int] = mapped_column(Integer, nullable=False)
+    updated_at: Mapped[int] = mapped_column(Integer, nullable=False)
+
+    # 关系
+    ssh_key: Mapped["SshKey"] = relationship("SshKey", back_populates="servers")
+    repos: Mapped[List["Repo"]] = relationship("Repo", back_populates="server", cascade="all, delete-orphan")
+
+    def __repr__(self) -> str:
+        return f"<Server(id={self.id}, name='{self.name}', url='{self.url}', status='{self.status}')>"
diff --git a/backend/app/models/ssh_key.py b/backend/app/models/ssh_key.py
new file mode 100644
index 0000000..2aac701
--- /dev/null
+++ b/backend/app/models/ssh_key.py
@@ -0,0 +1,28 @@
+"""
+SSH 密钥 ORM 模型.
+"""
+from sqlalchemy import String, Integer, Text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from app.database import Base
+from typing import Optional, List
+
+
+class SshKey(Base):
+    """
+    SSH 密钥模型.
+
+    存储加密的 SSH 私钥，用于 Git 操作.
+    """
+    __tablename__ = "ssh_keys"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(100), nullable=False, unique=True)
+    private_key: Mapped[str] = mapped_column(Text, nullable=False)
+    fingerprint: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    created_at: Mapped[int] = mapped_column(Integer, nullable=False)
+
+    # 关系
+    servers: Mapped[List["Server"]] = relationship("Server", back_populates="ssh_key")
+
+    def __repr__(self) -> str:
+        return f"<SshKey(id={self.id}, name='{self.name}', fingerprint='{self.fingerprint}')>"
diff --git a/backend/app/models/sync_log.py b/backend/app/models/sync_log.py
new file mode 100644
index 0000000..9d0a180
--- /dev/null
+++ b/backend/app/models/sync_log.py
@@ -0,0 +1,31 @@
+"""
+同步日志 ORM 模型.
+"""
+from sqlalchemy import String, Integer, Text, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from app.database import Base
+from typing import Optional, List
+
+
+class SyncLog(Base):
+    """
+    同步日志模型.
+
+    记录仓库同步操作的详细信息.
+    """
+    __tablename__ = "sync_logs"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    repo_id: Mapped[int] = mapped_column(Integer, ForeignKey("repos.id"), nullable=False)
+    status: Mapped[str] = mapped_column(String(20), nullable=False)
+    started_at: Mapped[int] = mapped_column(Integer, nullable=False)
+    finished_at: Mapped[int] = mapped_column(Integer, nullable=False)
+    commits_count: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    error_msg: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
+    created_at: Mapped[int] = mapped_column(Integer, nullable=False)
+
+    # 关系
+    repo: Mapped["Repo"] = relationship("Repo", back_populates="sync_logs")
+
+    def __repr__(self) -> str:
+        return f"<SyncLog(id={self.id}, repo_id={self.repo_id}, status='{self.status}', commits_count={self.commits_count})>"
diff --git a/backend/tests/test_models/__init__.py b/backend/tests/test_models/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/backend/tests/test_models/test_orm_models.py b/backend/tests/test_models/test_orm_models.py
new file mode 100644
index 0000000..6cfa8d2
--- /dev/null
+++ b/backend/tests/test_models/test_orm_models.py
@@ -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"