feat: migrate ORM base to sqlmodel-ext, add file viewers and WOPI integration

- Migrate SQLModel base classes, mixins, and database management to
  external sqlmodel-ext package; remove sqlmodels/base/, sqlmodels/mixin/,
  and sqlmodels/database.py
- Add file viewer/editor system with WOPI protocol support for
  collaborative editing (OnlyOffice, Collabora)
- Add enterprise edition license verification module (ee/)
- Add Dockerfile multi-stage build with Cython compilation support
- Add new dependencies: sqlmodel-ext, cryptography, whatthepatch

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-14 14:23:17 +08:00
parent 53b757de7a
commit eac0766e79
74 changed files with 4819 additions and 4837 deletions

View File

@@ -0,0 +1,178 @@
"""
文本文件 patch 逻辑单元测试
测试 whatthepatch 库的 patch 解析与应用,
以及换行符规范化和 SHA-256 哈希计算。
"""
import hashlib
import pytest
import whatthepatch
from whatthepatch.exceptions import HunkApplyException
class TestPatchApply:
"""测试 patch 解析与应用"""
def test_normal_patch(self) -> None:
"""正常 patch 应用"""
original = "line1\nline2\nline3"
patch_text = (
"--- a\n"
"+++ b\n"
"@@ -1,3 +1,3 @@\n"
" line1\n"
"-line2\n"
"+LINE2\n"
" line3\n"
)
diffs = list(whatthepatch.parse_patch(patch_text))
assert len(diffs) == 1
result = whatthepatch.apply_diff(diffs[0], original)
new_text = '\n'.join(result)
assert "LINE2" in new_text
assert "line2" not in new_text
def test_add_lines_patch(self) -> None:
"""添加行的 patch"""
original = "line1\nline2"
patch_text = (
"--- a\n"
"+++ b\n"
"@@ -1,2 +1,3 @@\n"
" line1\n"
" line2\n"
"+line3\n"
)
diffs = list(whatthepatch.parse_patch(patch_text))
result = whatthepatch.apply_diff(diffs[0], original)
new_text = '\n'.join(result)
assert "line3" in new_text
def test_delete_lines_patch(self) -> None:
"""删除行的 patch"""
original = "line1\nline2\nline3"
patch_text = (
"--- a\n"
"+++ b\n"
"@@ -1,3 +1,2 @@\n"
" line1\n"
"-line2\n"
" line3\n"
)
diffs = list(whatthepatch.parse_patch(patch_text))
result = whatthepatch.apply_diff(diffs[0], original)
new_text = '\n'.join(result)
assert "line2" not in new_text
assert "line1" in new_text
assert "line3" in new_text
def test_invalid_patch_format(self) -> None:
"""无效的 patch 格式返回空列表"""
diffs = list(whatthepatch.parse_patch("this is not a patch"))
assert len(diffs) == 0
def test_patch_context_mismatch(self) -> None:
"""patch 上下文不匹配时抛出异常"""
original = "line1\nline2\nline3\n"
patch_text = (
"--- a\n"
"+++ b\n"
"@@ -1,3 +1,3 @@\n"
" line1\n"
"-WRONG\n"
"+REPLACED\n"
" line3\n"
)
diffs = list(whatthepatch.parse_patch(patch_text))
with pytest.raises(HunkApplyException):
whatthepatch.apply_diff(diffs[0], original)
def test_empty_file_patch(self) -> None:
"""空文件应用 patch"""
original = ""
patch_text = (
"--- a\n"
"+++ b\n"
"@@ -0,0 +1,2 @@\n"
"+line1\n"
"+line2\n"
)
diffs = list(whatthepatch.parse_patch(patch_text))
result = whatthepatch.apply_diff(diffs[0], original)
new_text = '\n'.join(result)
assert "line1" in new_text
assert "line2" in new_text
class TestHashComputation:
"""测试 SHA-256 哈希计算"""
def test_hash_consistency(self) -> None:
"""相同内容产生相同哈希"""
content = "hello world\n"
content_bytes = content.encode('utf-8')
hash1 = hashlib.sha256(content_bytes).hexdigest()
hash2 = hashlib.sha256(content_bytes).hexdigest()
assert hash1 == hash2
assert len(hash1) == 64
def test_hash_differs_for_different_content(self) -> None:
"""不同内容产生不同哈希"""
hash1 = hashlib.sha256(b"content A").hexdigest()
hash2 = hashlib.sha256(b"content B").hexdigest()
assert hash1 != hash2
def test_hash_after_normalization(self) -> None:
"""换行符规范化后的哈希一致性"""
content_crlf = "line1\r\nline2\r\n"
content_lf = "line1\nline2\n"
# 规范化后应相同
normalized = content_crlf.replace('\r\n', '\n').replace('\r', '\n')
assert normalized == content_lf
hash_normalized = hashlib.sha256(normalized.encode('utf-8')).hexdigest()
hash_lf = hashlib.sha256(content_lf.encode('utf-8')).hexdigest()
assert hash_normalized == hash_lf
class TestLineEndingNormalization:
"""测试换行符规范化"""
def test_crlf_to_lf(self) -> None:
"""CRLF 转换为 LF"""
content = "line1\r\nline2\r\n"
normalized = content.replace('\r\n', '\n').replace('\r', '\n')
assert normalized == "line1\nline2\n"
def test_cr_to_lf(self) -> None:
"""CR 转换为 LF"""
content = "line1\rline2\r"
normalized = content.replace('\r\n', '\n').replace('\r', '\n')
assert normalized == "line1\nline2\n"
def test_lf_unchanged(self) -> None:
"""LF 保持不变"""
content = "line1\nline2\n"
normalized = content.replace('\r\n', '\n').replace('\r', '\n')
assert normalized == content
def test_mixed_line_endings(self) -> None:
"""混合换行符统一为 LF"""
content = "line1\r\nline2\rline3\n"
normalized = content.replace('\r\n', '\n').replace('\r', '\n')
assert normalized == "line1\nline2\nline3\n"

View File

@@ -0,0 +1,77 @@
"""
WOPI Token 单元测试
测试 WOPI 访问令牌的生成和验证。
"""
from uuid import uuid4
import pytest
import utils.JWT as JWT
from utils.JWT.wopi_token import create_wopi_token, verify_wopi_token
# 确保测试 secret key
JWT.SECRET_KEY = "test_secret_key_for_jwt_token_generation"
class TestWopiToken:
"""WOPI Token 测试"""
def test_create_and_verify_token(self) -> None:
"""创建和验证令牌"""
file_id = uuid4()
user_id = uuid4()
token, ttl = create_wopi_token(file_id, user_id, can_write=True)
assert isinstance(token, str)
assert isinstance(ttl, int)
assert ttl > 0
payload = verify_wopi_token(token)
assert payload is not None
assert payload.file_id == file_id
assert payload.user_id == user_id
assert payload.can_write is True
def test_verify_read_only_token(self) -> None:
"""验证只读令牌"""
file_id = uuid4()
user_id = uuid4()
token, ttl = create_wopi_token(file_id, user_id, can_write=False)
payload = verify_wopi_token(token)
assert payload is not None
assert payload.can_write is False
def test_verify_invalid_token(self) -> None:
"""验证无效令牌返回 None"""
payload = verify_wopi_token("invalid_token_string")
assert payload is None
def test_verify_non_wopi_token(self) -> None:
"""验证非 WOPI 类型令牌返回 None"""
import jwt as pyjwt
# 创建一个不含 type=wopi 的令牌
token = pyjwt.encode(
{"file_id": str(uuid4()), "user_id": str(uuid4()), "type": "download"},
JWT.SECRET_KEY,
algorithm="HS256",
)
payload = verify_wopi_token(token)
assert payload is None
def test_ttl_is_future_milliseconds(self) -> None:
"""TTL 应为未来的毫秒时间戳"""
import time
file_id = uuid4()
user_id = uuid4()
token, ttl = create_wopi_token(file_id, user_id)
current_ms = int(time.time() * 1000)
# TTL 应大于当前时间
assert ttl > current_ms
# TTL 不应超过 11 小时后10h + 余量)
assert ttl < current_ms + 11 * 3600 * 1000