Compare commits
No commits in common. 'main' and 'dev' have entirely different histories.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
Before Width: | Height: | Size: 194 KiB |
|
Before Width: | Height: | Size: 97 KiB |
|
Before Width: | Height: | Size: 213 KiB |
|
Before Width: | Height: | Size: 45 KiB |
|
Before Width: | Height: | Size: 232 KiB |
|
Before Width: | Height: | Size: 78 KiB |
|
Before Width: | Height: | Size: 124 KiB |
|
Before Width: | Height: | Size: 67 KiB |
|
Before Width: | Height: | Size: 499 KiB |
|
Before Width: | Height: | Size: 160 KiB |
|
Before Width: | Height: | Size: 72 KiB |
|
Before Width: | Height: | Size: 73 KiB |
|
Before Width: | Height: | Size: 72 KiB |
|
Before Width: | Height: | Size: 36 KiB |
|
Before Width: | Height: | Size: 56 KiB |
|
Before Width: | Height: | Size: 39 KiB |
|
Before Width: | Height: | Size: 19 KiB |
|
Before Width: | Height: | Size: 56 KiB |
|
Before Width: | Height: | Size: 56 KiB |
|
Before Width: | Height: | Size: 43 KiB |
|
Before Width: | Height: | Size: 501 KiB |
|
Before Width: | Height: | Size: 437 KiB |
|
Before Width: | Height: | Size: 311 KiB |
|
Before Width: | Height: | Size: 21 KiB |
|
Before Width: | Height: | Size: 33 KiB |
|
Before Width: | Height: | Size: 128 KiB |
|
Before Width: | Height: | Size: 126 KiB |
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
Before Width: | Height: | Size: 4.4 MiB |
@ -1,30 +0,0 @@
|
||||
[run]
|
||||
source = .
|
||||
omit =
|
||||
*/tests/*
|
||||
*/test_*.py
|
||||
*/__pycache__/*
|
||||
*/venv/*
|
||||
*/env/*
|
||||
setup.py
|
||||
*/site-packages/*
|
||||
|
||||
[report]
|
||||
exclude_lines =
|
||||
pragma: no cover
|
||||
def __repr__
|
||||
raise AssertionError
|
||||
raise NotImplementedError
|
||||
if __name__ == .__main__.:
|
||||
if TYPE_CHECKING:
|
||||
@abstractmethod
|
||||
@abc.abstractmethod
|
||||
|
||||
precision = 2
|
||||
show_missing = True
|
||||
|
||||
[html]
|
||||
directory = tests/coverage_html
|
||||
|
||||
[xml]
|
||||
output = coverage.xml
|
||||
@ -1,235 +0,0 @@
|
||||
import argparse
|
||||
import json
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
import ijson
|
||||
|
||||
|
||||
DB_PATH = Path("backend/instance/app.sqlite")
|
||||
|
||||
|
||||
def ensure_db_exists() -> None:
|
||||
if not DB_PATH.parent.exists():
|
||||
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
if not DB_PATH.exists():
|
||||
raise FileNotFoundError(
|
||||
f"Database not found at {DB_PATH}. Please run backend/scripts/init_db.py first."
|
||||
)
|
||||
|
||||
|
||||
def insert_user_if_needed(conn: sqlite3.Connection, uid: int) -> None:
|
||||
if uid is None:
|
||||
return
|
||||
conn.execute(
|
||||
"INSERT OR IGNORE INTO users(uid) VALUES (?)",
|
||||
(uid,),
|
||||
)
|
||||
|
||||
|
||||
def ingest_records(conn: sqlite3.Connection, record_path: Path, batch_size: int = 1000) -> int:
|
||||
count = 0
|
||||
print(f"Opening {record_path} for records import...")
|
||||
with record_path.open("rb") as fp:
|
||||
items = ijson.items(fp, "item")
|
||||
buffer = []
|
||||
for obj in items:
|
||||
submission_oid = (obj.get("_id") or {}).get("$oid")
|
||||
status = obj.get("status")
|
||||
uid = obj.get("uid")
|
||||
code = obj.get("code")
|
||||
lang = obj.get("lang")
|
||||
pid_int = obj.get("pid")
|
||||
domain_id = obj.get("domainId")
|
||||
score = obj.get("score")
|
||||
exec_time = obj.get("time")
|
||||
memory = obj.get("memory")
|
||||
judge_texts = json.dumps(obj.get("judgeTexts")) if obj.get("judgeTexts") is not None else None
|
||||
compiler_texts = json.dumps(obj.get("compilerTexts")) if obj.get("compilerTexts") is not None else None
|
||||
test_cases = json.dumps(obj.get("testCases")) if obj.get("testCases") is not None else None
|
||||
judge_at = (obj.get("judgeAt") or {}).get("$date")
|
||||
rejudged = 1 if obj.get("rejudged") else 0
|
||||
files_json = json.dumps(obj.get("files")) if obj.get("files") is not None else None
|
||||
subtasks_json = json.dumps(obj.get("subtasks")) if obj.get("subtasks") is not None else None
|
||||
ip = obj.get("ip")
|
||||
judger = obj.get("judger")
|
||||
|
||||
insert_user_if_needed(conn, uid)
|
||||
|
||||
buffer.append(
|
||||
(
|
||||
submission_oid,
|
||||
status,
|
||||
uid,
|
||||
code,
|
||||
lang,
|
||||
pid_int,
|
||||
domain_id,
|
||||
score,
|
||||
exec_time,
|
||||
memory,
|
||||
judge_texts,
|
||||
compiler_texts,
|
||||
test_cases,
|
||||
judge_at,
|
||||
rejudged,
|
||||
files_json,
|
||||
subtasks_json,
|
||||
ip,
|
||||
judger,
|
||||
)
|
||||
)
|
||||
|
||||
count += 1
|
||||
if count % 1000 == 0:
|
||||
print(f" Processed {count} records...")
|
||||
if len(buffer) >= batch_size:
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT OR REPLACE INTO submissions(
|
||||
submission_oid, status, uid, code, lang, pid_int, domain_id, score, exec_time, memory,
|
||||
judge_texts, compiler_texts, test_cases, judge_at, rejudged, files_json, subtasks_json, ip, judger
|
||||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||||
""",
|
||||
buffer,
|
||||
)
|
||||
conn.commit()
|
||||
print(f" Committed batch at {count} records")
|
||||
buffer.clear()
|
||||
|
||||
if buffer:
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT OR REPLACE INTO submissions(
|
||||
submission_oid, status, uid, code, lang, pid_int, domain_id, score, exec_time, memory,
|
||||
judge_texts, compiler_texts, test_cases, judge_at, rejudged, files_json, subtasks_json, ip, judger
|
||||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||||
""",
|
||||
buffer,
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
return count
|
||||
|
||||
|
||||
def ingest_documents(conn: sqlite3.Connection, document_path: Path, batch_size: int = 1000) -> int:
|
||||
count = 0
|
||||
print(f"Opening {document_path} for documents import...")
|
||||
with document_path.open("rb") as fp:
|
||||
items = ijson.items(fp, "item")
|
||||
buffer = []
|
||||
for obj in items:
|
||||
doc_oid = (obj.get("_id") or {}).get("$oid")
|
||||
doc_id = obj.get("docId")
|
||||
pid_code = obj.get("pid")
|
||||
title = obj.get("title")
|
||||
content_json = obj.get("content") # 原始即为 JSON 字符串
|
||||
owner = obj.get("owner")
|
||||
domain_id = obj.get("domainId")
|
||||
doc_type = obj.get("docType")
|
||||
tag_json = json.dumps(obj.get("tag")) if obj.get("tag") is not None else None
|
||||
hidden = 1 if obj.get("hidden") else 0
|
||||
n_submit = obj.get("nSubmit")
|
||||
n_accept = obj.get("nAccept")
|
||||
sort = obj.get("sort")
|
||||
data_json = json.dumps(obj.get("data")) if obj.get("data") is not None else None
|
||||
additional_file_json = json.dumps(obj.get("additional_file")) if obj.get("additional_file") is not None else None
|
||||
config = obj.get("config")
|
||||
if isinstance(config, dict):
|
||||
config = json.dumps(config)
|
||||
stats_json = json.dumps(obj.get("stats")) if obj.get("stats") is not None else None
|
||||
|
||||
buffer.append(
|
||||
(
|
||||
doc_oid,
|
||||
doc_id,
|
||||
pid_code,
|
||||
title,
|
||||
content_json,
|
||||
owner,
|
||||
domain_id,
|
||||
doc_type,
|
||||
tag_json,
|
||||
hidden,
|
||||
n_submit,
|
||||
n_accept,
|
||||
sort,
|
||||
data_json,
|
||||
additional_file_json,
|
||||
config,
|
||||
stats_json,
|
||||
)
|
||||
)
|
||||
|
||||
count += 1
|
||||
if count % 100 == 0:
|
||||
print(f" Processed {count} documents...")
|
||||
if len(buffer) >= batch_size:
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT OR REPLACE INTO problems(
|
||||
doc_oid, doc_id, pid_code, title, content_json, owner, domain_id, doc_type,
|
||||
tag_json, hidden, n_submit, n_accept, sort, data_json, additional_file_json, config, stats_json
|
||||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||||
""",
|
||||
buffer,
|
||||
)
|
||||
conn.commit()
|
||||
print(f" Committed documents batch at {count} records")
|
||||
buffer.clear()
|
||||
|
||||
if buffer:
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT OR REPLACE INTO problems(
|
||||
doc_oid, doc_id, pid_code, title, content_json, owner, domain_id, doc_type,
|
||||
tag_json, hidden, n_submit, n_accept, sort, data_json, additional_file_json, config, stats_json
|
||||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||||
""",
|
||||
buffer,
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
return count
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--record", type=Path, default=Path("data/record.json"))
|
||||
parser.add_argument("--document", type=Path, default=Path("data/document.json"))
|
||||
args = parser.parse_args()
|
||||
|
||||
ensure_db_exists()
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
try:
|
||||
conn.execute("PRAGMA journal_mode=WAL;")
|
||||
conn.execute("PRAGMA synchronous=NORMAL;")
|
||||
|
||||
print(f"Starting import...")
|
||||
print(f"Document file exists: {args.document.exists()}")
|
||||
print(f"Record file exists: {args.record.exists()}")
|
||||
|
||||
if args.document.exists():
|
||||
print(f"Importing documents from {args.document}...")
|
||||
n_doc = ingest_documents(conn, args.document)
|
||||
print(f"Imported documents: {n_doc}")
|
||||
else:
|
||||
print(f"Skip: document file not found: {args.document}")
|
||||
|
||||
if args.record.exists():
|
||||
print(f"Importing records from {args.record}...")
|
||||
n_rec = ingest_records(conn, args.record)
|
||||
print(f"Imported records: {n_rec}")
|
||||
else:
|
||||
print(f"Skip: record file not found: {args.record}")
|
||||
|
||||
print("Import completed!")
|
||||
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
|
||||
@ -1,51 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
配置文件
|
||||
"""
|
||||
import os
|
||||
|
||||
|
||||
class Config:
|
||||
"""基础配置"""
|
||||
# Flask配置
|
||||
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key-for-programming-evaluation-system'
|
||||
|
||||
# 数据库配置
|
||||
DB_PATH = 'backend/instance/app.sqlite'
|
||||
|
||||
# 应用配置
|
||||
DEBUG = True
|
||||
HOST = '0.0.0.0'
|
||||
PORT = 5000
|
||||
|
||||
# 登录配置
|
||||
DEFAULT_PASSWORD = '123456'
|
||||
|
||||
|
||||
class DevelopmentConfig(Config):
|
||||
"""开发环境配置"""
|
||||
DEBUG = True
|
||||
TESTING = False
|
||||
|
||||
|
||||
class ProductionConfig(Config):
|
||||
"""生产环境配置"""
|
||||
DEBUG = False
|
||||
TESTING = False
|
||||
|
||||
|
||||
class TestingConfig(Config):
|
||||
"""测试环境配置"""
|
||||
DEBUG = True
|
||||
TESTING = True
|
||||
|
||||
|
||||
# 配置字典
|
||||
config = {
|
||||
'development': DevelopmentConfig,
|
||||
'production': ProductionConfig,
|
||||
'testing': TestingConfig,
|
||||
'default': DevelopmentConfig
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Binary file not shown.
@ -1,35 +0,0 @@
|
||||
[pytest]
|
||||
# Pytest配置文件
|
||||
testpaths = tests
|
||||
python_files = test_*.py
|
||||
python_classes = Test*
|
||||
python_functions = test_*
|
||||
|
||||
# 输出选项
|
||||
addopts =
|
||||
-v
|
||||
--strict-markers
|
||||
--tb=short
|
||||
--cov=.
|
||||
--cov-report=html:tests/coverage_html
|
||||
--cov-report=xml:coverage.xml
|
||||
--cov-report=term-missing
|
||||
--cov-report=json:tests/coverage.json
|
||||
--cov-config=.coveragerc
|
||||
--junitxml=tests/test-results.xml
|
||||
|
||||
# 标记定义
|
||||
markers =
|
||||
unit: 单元测试
|
||||
integration: 集成测试
|
||||
slow: 慢速测试
|
||||
auth: 认证相关测试
|
||||
stats: 统计相关测试
|
||||
assessment: 评估相关测试
|
||||
api: API测试
|
||||
lstm: LSTM模型测试
|
||||
|
||||
# 覆盖率配置
|
||||
filterwarnings =
|
||||
ignore::DeprecationWarning
|
||||
ignore::PendingDeprecationWarning
|
||||
@ -1,39 +0,0 @@
|
||||
Flask==2.3.3
|
||||
Werkzeug==2.3.7
|
||||
Jinja2==3.1.3
|
||||
MarkupSafe==2.1.5
|
||||
itsdangerous==2.1.2
|
||||
click==8.1.7
|
||||
|
||||
flask-sqlalchemy==3.1.1
|
||||
SQLAlchemy==2.0.20
|
||||
flask-migrate==4.0.4
|
||||
|
||||
flask-login==0.6.2
|
||||
flask-wtf==1.2.1
|
||||
flask-cors==4.0.0
|
||||
email-validator==2.0.0
|
||||
|
||||
pandas==2.2.3
|
||||
numpy==1.26.4
|
||||
matplotlib==3.8.3
|
||||
seaborn==0.13.2
|
||||
scikit-learn==1.4.1.post1
|
||||
python-dateutil==2.9.0.post0
|
||||
six==1.17.0
|
||||
|
||||
|
||||
torch>=2.2.0
|
||||
torchvision>=0.17.0
|
||||
torchaudio>=2.2.0
|
||||
|
||||
|
||||
python-dotenv==1.0.0
|
||||
gunicorn==21.2.0
|
||||
pytest==7.3.1
|
||||
pytest-cov==4.1.0
|
||||
pytest-flask==1.2.0
|
||||
coverage==7.3.0
|
||||
ijson==3.2.3
|
||||
scikit-learn==1.4.1.post1
|
||||
requests==2.31.0
|
||||
@ -1,47 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
测试运行脚本
|
||||
快速运行所有测试并生成覆盖率报告
|
||||
"""
|
||||
import subprocess
|
||||
import sys
|
||||
import os
|
||||
|
||||
|
||||
def run_tests():
|
||||
"""运行测试"""
|
||||
print("=" * 60)
|
||||
print("开始运行测试...")
|
||||
print("=" * 60)
|
||||
|
||||
# 运行pytest
|
||||
cmd = [
|
||||
sys.executable, '-m', 'pytest',
|
||||
'tests/',
|
||||
'-v',
|
||||
'--cov=.',
|
||||
'--cov-report=html:tests/coverage_html',
|
||||
'--cov-report=xml:coverage.xml',
|
||||
'--cov-report=term-missing',
|
||||
'--cov-report=json:tests/coverage.json',
|
||||
'--junitxml=tests/test-results.xml',
|
||||
'--tb=short'
|
||||
]
|
||||
|
||||
result = subprocess.run(cmd, cwd=os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
if result.returncode == 0:
|
||||
print("✓ 测试完成！")
|
||||
print("✓ 覆盖率报告已生成到: tests/coverage_html/index.html")
|
||||
else:
|
||||
print("✗ 部分测试失败")
|
||||
print(" 请查看上方输出了解详情")
|
||||
print("=" * 60)
|
||||
|
||||
return result.returncode
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(run_tests())
|
||||
@ -1,20 +0,0 @@
|
||||
"""
|
||||
服务层模块
|
||||
包含所有业务逻辑
|
||||
"""
|
||||
|
||||
from .auth_service import AuthService, User
|
||||
from .stats_service import StatsService
|
||||
from .assessment_service import AssessmentService
|
||||
from .suggestion_service import SuggestionService
|
||||
from .api_service import ApiService
|
||||
|
||||
__all__ = [
|
||||
'AuthService',
|
||||
'User',
|
||||
'StatsService',
|
||||
'AssessmentService',
|
||||
'SuggestionService',
|
||||
'ApiService'
|
||||
]
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -1,22 +0,0 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="zh-CN">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
<title>{% block title %}编程能力个性化评价系统{% endblock %}</title>
|
||||
|
||||
<!-- Bootstrap CSS -->
|
||||
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
|
||||
<!-- Font Awesome -->
|
||||
<link href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.4.0/css/all.min.css" rel="stylesheet">
|
||||
|
||||
{% block styles %}{% endblock %}
|
||||
</head>
|
||||
<body>
|
||||
{% block content %}{% endblock %}
|
||||
|
||||
<!-- Bootstrap JS -->
|
||||
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script>
|
||||
{% block scripts %}{% endblock %}
|
||||
</body>
|
||||
</html>
|
||||
File diff suppressed because it is too large
Load Diff
@ -1,5 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
测试包初始化文件
|
||||
"""
|
||||
@ -1,178 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Pytest配置文件
|
||||
提供公共的fixtures和测试配置
|
||||
"""
|
||||
import pytest
|
||||
import sqlite3
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from flask import Flask
|
||||
|
||||
# 添加项目根目录到Python路径
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
||||
|
||||
from services import AuthService, StatsService, AssessmentService, SuggestionService
|
||||
from data_analyzer import DataAnalyzer
|
||||
|
||||
|
||||
@pytest.fixture(scope='session')
|
||||
def test_db_path():
|
||||
"""创建临时测试数据库路径"""
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
db_path = os.path.join(temp_dir, 'test.sqlite')
|
||||
yield db_path
|
||||
# 清理测试数据库
|
||||
if os.path.exists(db_path):
|
||||
os.remove(db_path)
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def init_test_db(test_db_path):
|
||||
"""初始化测试数据库"""
|
||||
conn = sqlite3.connect(test_db_path)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 创建用户表
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
uid INTEGER PRIMARY KEY,
|
||||
username TEXT
|
||||
)
|
||||
""")
|
||||
|
||||
# 创建提交表
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS submissions (
|
||||
sid INTEGER PRIMARY KEY,
|
||||
uid INTEGER,
|
||||
pid TEXT,
|
||||
pid_int INTEGER,
|
||||
status INTEGER,
|
||||
score INTEGER,
|
||||
lang TEXT,
|
||||
submit_time TEXT,
|
||||
FOREIGN KEY (uid) REFERENCES users(uid)
|
||||
)
|
||||
""")
|
||||
|
||||
# 创建问题表
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS problems (
|
||||
pid INTEGER PRIMARY KEY,
|
||||
title TEXT,
|
||||
difficulty TEXT,
|
||||
tags TEXT
|
||||
)
|
||||
""")
|
||||
|
||||
# 插入测试用户
|
||||
cursor.executemany(
|
||||
"INSERT INTO users (uid, username) VALUES (?, ?)",
|
||||
[(1, '测试用户1'), (2, '测试用户2'), (3, '测试用户3')]
|
||||
)
|
||||
|
||||
# 插入测试提交数据
|
||||
test_submissions = [
|
||||
(1, 1, '1001', 1001, 1, 100, 'Python', '2024-01-01 10:00:00'),
|
||||
(2, 1, '1002', 1002, 1, 95, 'Python', '2024-01-02 11:00:00'),
|
||||
(3, 1, '1003', 1003, 0, 0, 'Java', '2024-01-03 12:00:00'),
|
||||
(4, 2, '1001', 1001, 1, 90, 'C++', '2024-01-01 13:00:00'),
|
||||
(5, 2, '1002', 1002, 3, 85, 'Python', '2024-01-02 14:00:00'),
|
||||
]
|
||||
cursor.executemany(
|
||||
"INSERT INTO submissions (sid, uid, pid, pid_int, status, score, lang, submit_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
test_submissions
|
||||
)
|
||||
|
||||
# 插入测试问题数据
|
||||
test_problems = [
|
||||
(1001, '两数之和', 'Easy', 'Array,Hash Table'),
|
||||
(1002, '二叉树遍历', 'Medium', 'Tree,DFS'),
|
||||
(1003, '动态规划问题', 'Hard', 'DP,Math'),
|
||||
]
|
||||
cursor.executemany(
|
||||
"INSERT INTO problems (pid, title, difficulty, tags) VALUES (?, ?, ?, ?)",
|
||||
test_problems
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
yield test_db_path
|
||||
|
||||
# 测试后清理数据
|
||||
conn = sqlite3.connect(test_db_path)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("DELETE FROM submissions")
|
||||
cursor.execute("DELETE FROM users")
|
||||
cursor.execute("DELETE FROM problems")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_service(init_test_db):
|
||||
"""创建认证服务实例"""
|
||||
return AuthService(init_test_db)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stats_service(init_test_db):
|
||||
"""创建统计服务实例"""
|
||||
return StatsService(init_test_db)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def assessment_service(init_test_db):
|
||||
"""创建评估服务实例"""
|
||||
return AssessmentService(init_test_db)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def suggestion_service(init_test_db):
|
||||
"""创建建议服务实例"""
|
||||
return SuggestionService(init_test_db)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def data_analyzer(init_test_db):
|
||||
"""创建数据分析器实例"""
|
||||
return DataAnalyzer(init_test_db)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app(init_test_db):
|
||||
"""创建Flask测试应用"""
|
||||
from app import app as flask_app
|
||||
flask_app.config['TESTING'] = True
|
||||
flask_app.config['DB_PATH'] = init_test_db
|
||||
flask_app.config['WTF_CSRF_ENABLED'] = False
|
||||
return flask_app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(app):
|
||||
"""创建Flask测试客户端"""
|
||||
return app.test_client()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def runner(app):
|
||||
"""创建Flask CLI测试运行器"""
|
||||
return app.test_cli_runner()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def authenticated_client(client, init_test_db):
|
||||
"""创建已登录的测试客户端"""
|
||||
# 登录用户
|
||||
client.post('/login', data={
|
||||
'username': '测试用户1',
|
||||
'password': '123456'
|
||||
}, follow_redirects=True)
|
||||
yield client
|
||||
# 登出
|
||||
client.get('/logout', follow_redirects=True)
|
||||
File diff suppressed because one or more lines are too long
@ -1,87 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
数据库集成测试
|
||||
"""
|
||||
import pytest
|
||||
import sqlite3
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestDatabaseIntegration:
|
||||
"""数据库集成测试"""
|
||||
|
||||
def test_database_connection(self, init_test_db):
|
||||
"""测试数据库连接"""
|
||||
conn = sqlite3.connect(init_test_db)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT 1")
|
||||
result = cursor.fetchone()
|
||||
conn.close()
|
||||
|
||||
assert result[0] == 1
|
||||
|
||||
def test_users_table_exists(self, init_test_db):
|
||||
"""测试用户表存在"""
|
||||
conn = sqlite3.connect(init_test_db)
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""
|
||||
SELECT name FROM sqlite_master
|
||||
WHERE type='table' AND name='users'
|
||||
""")
|
||||
result = cursor.fetchone()
|
||||
conn.close()
|
||||
|
||||
assert result is not None
|
||||
assert result[0] == 'users'
|
||||
|
||||
def test_submissions_table_exists(self, init_test_db):
|
||||
"""测试提交表存在"""
|
||||
conn = sqlite3.connect(init_test_db)
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""
|
||||
SELECT name FROM sqlite_master
|
||||
WHERE type='table' AND name='submissions'
|
||||
""")
|
||||
result = cursor.fetchone()
|
||||
conn.close()
|
||||
|
||||
assert result is not None
|
||||
|
||||
def test_data_integrity(self, init_test_db):
|
||||
"""测试数据完整性"""
|
||||
conn = sqlite3.connect(init_test_db)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 检查用户数据
|
||||
cursor.execute("SELECT COUNT(*) FROM users")
|
||||
user_count = cursor.fetchone()[0]
|
||||
assert user_count >= 3
|
||||
|
||||
# 检查提交数据
|
||||
cursor.execute("SELECT COUNT(*) FROM submissions")
|
||||
submission_count = cursor.fetchone()[0]
|
||||
assert submission_count >= 5
|
||||
|
||||
conn.close()
|
||||
|
||||
def test_foreign_key_constraint(self, init_test_db):
|
||||
"""测试外键约束"""
|
||||
conn = sqlite3.connect(init_test_db)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 查询提交记录的用户是否存在
|
||||
cursor.execute("""
|
||||
SELECT s.uid, u.uid
|
||||
FROM submissions s
|
||||
LEFT JOIN users u ON s.uid = u.uid
|
||||
WHERE s.uid IS NOT NULL
|
||||
""")
|
||||
results = cursor.fetchall()
|
||||
|
||||
for submission_uid, user_uid in results:
|
||||
assert user_uid is not None, f"用户 {submission_uid} 不存在"
|
||||
|
||||
conn.close()
|
||||
File diff suppressed because one or more lines are too long
@ -1,126 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
评估服务单元测试
|
||||
"""
|
||||
import pytest
|
||||
from services.assessment_service import AssessmentService
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.assessment
|
||||
class TestAssessmentService:
|
||||
"""评估服务测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_stats_data(self):
|
||||
"""模拟统计数据"""
|
||||
return {
|
||||
'total_submissions': 10,
|
||||
'accepted_submissions': 7,
|
||||
'acceptance_rate': 70.0,
|
||||
'avg_score': 85,
|
||||
'max_score': 100,
|
||||
'unique_problems': 5,
|
||||
'rank': 10
|
||||
}
|
||||
|
||||
def test_get_user_assessment_structure(self, assessment_service, mock_stats_data):
|
||||
"""测试评估结果结构"""
|
||||
assessment = assessment_service.get_user_assessment(1, mock_stats_data)
|
||||
|
||||
assert assessment is not None
|
||||
assert 'overall_score' in assessment
|
||||
assert 'level' in assessment
|
||||
assert 'total_problems_solved' in assessment
|
||||
assert 'acceptance_rate' in assessment
|
||||
assert 'strengths' in assessment
|
||||
assert 'improvement_areas' in assessment
|
||||
|
||||
def test_overall_score_calculation(self, assessment_service, mock_stats_data):
|
||||
"""测试综合评分计算"""
|
||||
assessment = assessment_service.get_user_assessment(1, mock_stats_data)
|
||||
|
||||
score = assessment['overall_score']
|
||||
assert 0 <= score <= 100
|
||||
assert isinstance(score, (int, float))
|
||||
|
||||
def test_level_determination(self, assessment_service, mock_stats_data):
|
||||
"""测试等级判定"""
|
||||
assessment = assessment_service.get_user_assessment(1, mock_stats_data)
|
||||
|
||||
level = assessment['level']
|
||||
valid_levels = ['初学者', '入门', '进阶', '熟练', '精通', '专家']
|
||||
assert level in valid_levels
|
||||
|
||||
def test_strengths_analysis(self, assessment_service, mock_stats_data):
|
||||
"""测试优势领域分析"""
|
||||
assessment = assessment_service.get_user_assessment(1, mock_stats_data)
|
||||
|
||||
strengths = assessment['strengths']
|
||||
assert isinstance(strengths, list)
|
||||
assert len(strengths) >= 0
|
||||
|
||||
def test_improvement_areas_analysis(self, assessment_service, mock_stats_data):
|
||||
"""测试改进领域分析"""
|
||||
assessment = assessment_service.get_user_assessment(1, mock_stats_data)
|
||||
|
||||
improvements = assessment['improvement_areas']
|
||||
assert isinstance(improvements, list)
|
||||
assert len(improvements) >= 0
|
||||
|
||||
def test_radar_data_generation(self, assessment_service, mock_stats_data):
|
||||
"""测试雷达图数据生成"""
|
||||
assessment = assessment_service.get_user_assessment(1, mock_stats_data)
|
||||
|
||||
if 'radar_data' in assessment:
|
||||
radar_data = assessment['radar_data']
|
||||
assert isinstance(radar_data, (list, dict))
|
||||
|
||||
def test_learning_suggestions(self, assessment_service, mock_stats_data):
|
||||
"""测试学习建议生成"""
|
||||
assessment = assessment_service.get_user_assessment(1, mock_stats_data)
|
||||
|
||||
if 'learning_suggestions' in assessment:
|
||||
suggestions = assessment['learning_suggestions']
|
||||
assert isinstance(suggestions, list)
|
||||
|
||||
def test_comparison_data(self, assessment_service, mock_stats_data):
|
||||
"""测试对比数据"""
|
||||
assessment = assessment_service.get_user_assessment(1, mock_stats_data)
|
||||
|
||||
if 'comparison' in assessment:
|
||||
comparison = assessment['comparison']
|
||||
assert 'personal_avg' in comparison or 'global_avg' in comparison
|
||||
|
||||
def test_assessment_with_low_stats(self, assessment_service):
|
||||
"""测试低统计数据的评估"""
|
||||
low_stats = {
|
||||
'total_submissions': 2,
|
||||
'accepted_submissions': 0,
|
||||
'acceptance_rate': 0,
|
||||
'avg_score': 0,
|
||||
'max_score': 0,
|
||||
'unique_problems': 0,
|
||||
'rank': 999
|
||||
}
|
||||
|
||||
assessment = assessment_service.get_user_assessment(1, low_stats)
|
||||
assert assessment is not None
|
||||
assert assessment['overall_score'] >= 0
|
||||
|
||||
def test_assessment_with_high_stats(self, assessment_service):
|
||||
"""测试高统计数据的评估"""
|
||||
high_stats = {
|
||||
'total_submissions': 100,
|
||||
'accepted_submissions': 95,
|
||||
'acceptance_rate': 95.0,
|
||||
'avg_score': 98,
|
||||
'max_score': 100,
|
||||
'unique_problems': 80,
|
||||
'rank': 1
|
||||
}
|
||||
|
||||
assessment = assessment_service.get_user_assessment(1, high_stats)
|
||||
assert assessment is not None
|
||||
assert assessment['overall_score'] > 60
|
||||
@ -1,90 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
认证服务单元测试
|
||||
"""
|
||||
import pytest
|
||||
from services.auth_service import User, AuthService
|
||||
|
||||
|
||||
class TestUser:
|
||||
"""用户模型测试"""
|
||||
|
||||
def test_user_creation(self):
|
||||
"""测试用户创建"""
|
||||
user = User(uid=1, username='测试用户')
|
||||
assert user.uid == 1
|
||||
assert user.username == '测试用户'
|
||||
assert user.id == '1'
|
||||
|
||||
def test_user_default_username(self):
|
||||
"""测试默认用户名"""
|
||||
user = User(uid=2)
|
||||
assert user.username == '用户2'
|
||||
|
||||
def test_get_id(self):
|
||||
"""测试获取用户ID"""
|
||||
user = User(uid=123)
|
||||
assert user.get_id() == '123'
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.auth
|
||||
class TestAuthService:
|
||||
"""认证服务测试"""
|
||||
|
||||
def test_load_user_exists(self, auth_service):
|
||||
"""测试加载存在的用户"""
|
||||
user = auth_service.load_user('1')
|
||||
assert user is not None
|
||||
assert user.uid == 1
|
||||
assert user.username == '测试用户1'
|
||||
|
||||
def test_load_user_not_exists(self, auth_service):
|
||||
"""测试加载不存在的用户"""
|
||||
user = auth_service.load_user('999')
|
||||
assert user is None
|
||||
|
||||
def test_load_user_invalid_id(self, auth_service):
|
||||
"""测试加载无效用户ID"""
|
||||
user = auth_service.load_user('invalid')
|
||||
assert user is None
|
||||
|
||||
def test_authenticate_by_username(self, auth_service):
|
||||
"""测试通过用户名认证"""
|
||||
user = auth_service.authenticate_user('测试用户1', '123456')
|
||||
assert user is not None
|
||||
assert user.uid == 1
|
||||
assert user.username == '测试用户1'
|
||||
|
||||
def test_authenticate_by_uid(self, auth_service):
|
||||
"""测试通过UID认证"""
|
||||
user = auth_service.authenticate_user('1', '123456')
|
||||
assert user is not None
|
||||
assert user.uid == 1
|
||||
|
||||
def test_authenticate_wrong_password(self, auth_service):
|
||||
"""测试错误密码"""
|
||||
user = auth_service.authenticate_user('测试用户1', 'wrong_password')
|
||||
assert user is None
|
||||
|
||||
def test_authenticate_nonexistent_user(self, auth_service):
|
||||
"""测试认证不存在的用户"""
|
||||
user = auth_service.authenticate_user('nonexistent', '123456')
|
||||
assert user is None
|
||||
|
||||
def test_get_all_users(self, auth_service):
|
||||
"""测试获取所有用户"""
|
||||
users = auth_service.get_all_users()
|
||||
assert len(users) >= 3
|
||||
assert any(user['uid'] == 1 for user in users)
|
||||
assert any(user['username'] == '测试用户1' for user in users)
|
||||
|
||||
def test_get_all_users_empty_db(self, test_db_path):
|
||||
"""测试空数据库获取用户"""
|
||||
import os
|
||||
import tempfile
|
||||
temp_db = os.path.join(tempfile.mkdtemp(), 'empty.db')
|
||||
service = AuthService(temp_db)
|
||||
users = service.get_all_users()
|
||||
assert users == []
|
||||
@ -1,76 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
缓存管理器单元测试
|
||||
"""
|
||||
import pytest
|
||||
import time
|
||||
from services.cache_manager import CacheManager, get_cache_manager
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestCacheManager:
|
||||
"""缓存管理器测试"""
|
||||
|
||||
def test_cache_manager_singleton(self):
|
||||
"""测试单例模式"""
|
||||
cache1 = get_cache_manager()
|
||||
cache2 = get_cache_manager()
|
||||
assert cache1 is cache2
|
||||
|
||||
def test_set_and_get(self):
|
||||
"""测试设置和获取缓存"""
|
||||
cache = get_cache_manager()
|
||||
cache.set('test_key', 'test_value', ttl=10)
|
||||
|
||||
value = cache.get('test_key')
|
||||
assert value == 'test_value'
|
||||
|
||||
def test_get_nonexistent_key(self):
|
||||
"""测试获取不存在的键"""
|
||||
cache = get_cache_manager()
|
||||
value = cache.get('nonexistent_key')
|
||||
assert value is None
|
||||
|
||||
def test_cache_expiration(self):
|
||||
"""测试缓存过期"""
|
||||
cache = get_cache_manager()
|
||||
cache.set('expire_key', 'expire_value', ttl=1)
|
||||
|
||||
# 立即获取应该成功
|
||||
assert cache.get('expire_key') == 'expire_value'
|
||||
|
||||
# 等待过期
|
||||
time.sleep(2)
|
||||
assert cache.get('expire_key') is None
|
||||
|
||||
def test_delete_cache(self):
|
||||
"""测试删除缓存"""
|
||||
cache = get_cache_manager()
|
||||
cache.set('delete_key', 'delete_value')
|
||||
|
||||
cache.delete('delete_key')
|
||||
assert cache.get('delete_key') is None
|
||||
|
||||
def test_clear_cache(self):
|
||||
"""测试清空缓存"""
|
||||
cache = get_cache_manager()
|
||||
cache.set('key1', 'value1')
|
||||
cache.set('key2', 'value2')
|
||||
|
||||
cache.clear()
|
||||
assert cache.get('key1') is None
|
||||
assert cache.get('key2') is None
|
||||
|
||||
def test_cache_with_complex_data(self):
|
||||
"""测试缓存复杂数据"""
|
||||
cache = get_cache_manager()
|
||||
complex_data = {
|
||||
'list': [1, 2, 3],
|
||||
'dict': {'a': 1, 'b': 2},
|
||||
'nested': {'x': [1, 2], 'y': {'z': 3}}
|
||||
}
|
||||
|
||||
cache.set('complex_key', complex_data)
|
||||
retrieved = cache.get('complex_key')
|
||||
assert retrieved == complex_data
|
||||
@ -1,63 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
配置文件单元测试
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestConfig:
|
||||
"""配置测试"""
|
||||
|
||||
def test_config_import(self):
|
||||
"""测试配置导入"""
|
||||
from config import Config, DevelopmentConfig, ProductionConfig, TestingConfig
|
||||
|
||||
assert Config is not None
|
||||
assert DevelopmentConfig is not None
|
||||
assert ProductionConfig is not None
|
||||
assert TestingConfig is not None
|
||||
|
||||
def test_development_config(self):
|
||||
"""测试开发环境配置"""
|
||||
from config import DevelopmentConfig
|
||||
|
||||
assert DevelopmentConfig.DEBUG is True
|
||||
assert DevelopmentConfig.TESTING is False
|
||||
|
||||
def test_production_config(self):
|
||||
"""测试生产环境配置"""
|
||||
from config import ProductionConfig
|
||||
|
||||
assert ProductionConfig.DEBUG is False
|
||||
assert ProductionConfig.TESTING is False
|
||||
|
||||
def test_testing_config(self):
|
||||
"""测试测试环境配置"""
|
||||
from config import TestingConfig
|
||||
|
||||
assert TestingConfig.TESTING is True
|
||||
|
||||
def test_config_dict(self):
|
||||
"""测试配置字典"""
|
||||
from config import config
|
||||
|
||||
assert 'development' in config
|
||||
assert 'production' in config
|
||||
assert 'testing' in config
|
||||
assert 'default' in config
|
||||
|
||||
def test_secret_key_exists(self):
|
||||
"""测试密钥存在"""
|
||||
from config import Config
|
||||
|
||||
assert hasattr(Config, 'SECRET_KEY')
|
||||
assert Config.SECRET_KEY is not None
|
||||
|
||||
def test_db_path_exists(self):
|
||||
"""测试数据库路径"""
|
||||
from config import Config
|
||||
|
||||
assert hasattr(Config, 'DB_PATH')
|
||||
assert Config.DB_PATH is not None
|
||||
@ -1,62 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
数据分析器单元测试
|
||||
"""
|
||||
import pytest
|
||||
from data_analyzer import DataAnalyzer
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDataAnalyzer:
|
||||
"""数据分析器测试"""
|
||||
|
||||
def test_analyzer_creation(self, data_analyzer):
|
||||
"""测试分析器创建"""
|
||||
assert data_analyzer is not None
|
||||
assert hasattr(data_analyzer, 'db_path')
|
||||
|
||||
def test_analyze_user_stats(self, data_analyzer):
|
||||
"""测试分析用户统计"""
|
||||
try:
|
||||
stats = data_analyzer.analyze_user_stats(1)
|
||||
assert stats is not None
|
||||
assert isinstance(stats, dict)
|
||||
except Exception as e:
|
||||
pytest.skip(f"analyze_user_stats方法异常: {e}")
|
||||
|
||||
def test_get_difficulty_stats(self, data_analyzer):
|
||||
"""测试获取难度统计"""
|
||||
try:
|
||||
if hasattr(data_analyzer, 'get_difficulty_stats'):
|
||||
stats = data_analyzer.get_difficulty_stats(1)
|
||||
assert isinstance(stats, (list, dict))
|
||||
except Exception:
|
||||
pytest.skip("get_difficulty_stats方法不可用")
|
||||
|
||||
def test_get_time_trends(self, data_analyzer):
|
||||
"""测试获取时间趋势"""
|
||||
try:
|
||||
if hasattr(data_analyzer, 'get_time_trends'):
|
||||
trends = data_analyzer.get_time_trends(1)
|
||||
assert isinstance(trends, (list, dict))
|
||||
except Exception:
|
||||
pytest.skip("get_time_trends方法不可用")
|
||||
|
||||
def test_get_comparison_data(self, data_analyzer):
|
||||
"""测试获取对比数据"""
|
||||
try:
|
||||
comparison = data_analyzer.get_comparison_data(1)
|
||||
assert comparison is not None
|
||||
assert isinstance(comparison, dict)
|
||||
except Exception as e:
|
||||
pytest.skip(f"get_comparison_data方法异常: {e}")
|
||||
|
||||
def test_get_learning_insights(self, data_analyzer):
|
||||
"""测试获取学习洞察"""
|
||||
try:
|
||||
insights = data_analyzer.get_learning_insights(1)
|
||||
assert insights is not None
|
||||
assert isinstance(insights, list)
|
||||
except Exception as e:
|
||||
pytest.skip(f"get_learning_insights方法异常: {e}")
|
||||
@ -1,139 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
LSTM预测器单元测试
|
||||
"""
|
||||
import pytest
|
||||
import torch
|
||||
import numpy as np
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.lstm
|
||||
class TestLSTMPredictor:
|
||||
"""LSTM预测器测试"""
|
||||
|
||||
def test_lstm_predictor_import(self):
|
||||
"""测试LSTM预测器导入"""
|
||||
try:
|
||||
from services.lstm_predictor import LSTMPredictor
|
||||
assert LSTMPredictor is not None
|
||||
except ImportError:
|
||||
pytest.skip("LSTM预测器模块不存在")
|
||||
|
||||
def test_lstm_predictor_initialization(self):
|
||||
"""测试LSTM预测器初始化"""
|
||||
try:
|
||||
from services.lstm_predictor import LSTMPredictor
|
||||
predictor = LSTMPredictor(db_path='backend/instance/app.sqlite')
|
||||
assert predictor is not None
|
||||
except Exception as e:
|
||||
pytest.skip(f"LSTM预测器初始化失败: {e}")
|
||||
|
||||
def test_lstm_model_exists(self):
|
||||
"""测试LSTM模型文件存在"""
|
||||
import os
|
||||
model_path = 'models/lstm_knowledge_predictor.pth'
|
||||
|
||||
if os.path.exists(model_path):
|
||||
assert True
|
||||
else:
|
||||
pytest.skip("LSTM模型文件不存在")
|
||||
|
||||
def test_predict_knowledge_mastery(self):
|
||||
"""测试知识掌握度预测"""
|
||||
try:
|
||||
from services.lstm_predictor import LSTMPredictor
|
||||
predictor = LSTMPredictor(db_path='backend/instance/app.sqlite')
|
||||
|
||||
# 测试预测
|
||||
result = predictor.predict_knowledge_mastery(1)
|
||||
|
||||
if result:
|
||||
assert isinstance(result, dict)
|
||||
except Exception as e:
|
||||
pytest.skip(f"预测功能测试失败: {e}")
|
||||
|
||||
def test_predict_with_invalid_user(self):
|
||||
"""测试无效用户预测"""
|
||||
try:
|
||||
from services.lstm_predictor import LSTMPredictor
|
||||
predictor = LSTMPredictor(db_path='backend/instance/app.sqlite')
|
||||
|
||||
result = predictor.predict_knowledge_mastery(999999)
|
||||
# 应该返回None或空结果
|
||||
assert result is None or result == {}
|
||||
except Exception as e:
|
||||
pytest.skip(f"无效用户测试失败: {e}")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.lstm
|
||||
@pytest.mark.slow
|
||||
class TestLSTMModel:
|
||||
"""LSTM模型测试"""
|
||||
|
||||
def test_model_loading(self):
|
||||
"""测试模型加载"""
|
||||
try:
|
||||
import os
|
||||
model_path = 'models/lstm_knowledge_predictor.pth'
|
||||
|
||||
if not os.path.exists(model_path):
|
||||
pytest.skip("模型文件不存在")
|
||||
|
||||
model_data = torch.load(model_path, map_location='cpu')
|
||||
assert model_data is not None
|
||||
except Exception as e:
|
||||
pytest.skip(f"模型加载失败: {e}")
|
||||
|
||||
def test_model_inference(self):
|
||||
"""测试模型推理"""
|
||||
try:
|
||||
from services.lstm_predictor import LSTMPredictor
|
||||
import os
|
||||
|
||||
if not os.path.exists('models/lstm_knowledge_predictor.pth'):
|
||||
pytest.skip("模型文件不存在")
|
||||
|
||||
predictor = LSTMPredictor(db_path='backend/instance/app.sqlite')
|
||||
|
||||
# 创建测试输入
|
||||
test_input = torch.randn(1, 10, 5) # batch_size=1, seq_len=10, features=5
|
||||
|
||||
# 测试模型可以处理输入
|
||||
# 注意：这只是测试模型结构，不测试实际预测效果
|
||||
assert True
|
||||
except Exception as e:
|
||||
pytest.skip(f"模型推理测试失败: {e}")
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.lstm
|
||||
class TestLSTMIntegration:
|
||||
"""LSTM集成测试"""
|
||||
|
||||
def test_lstm_data_preparation(self):
|
||||
"""测试LSTM数据准备"""
|
||||
try:
|
||||
from services.lstm_predictor import LSTMPredictor
|
||||
predictor = LSTMPredictor(db_path='backend/instance/app.sqlite')
|
||||
|
||||
# 测试数据准备函数
|
||||
if hasattr(predictor, 'prepare_data'):
|
||||
data = predictor.prepare_data(1)
|
||||
assert data is not None
|
||||
except Exception as e:
|
||||
pytest.skip(f"数据准备测试失败: {e}")
|
||||
|
||||
def test_lstm_feature_extraction(self):
|
||||
"""测试LSTM特征提取"""
|
||||
try:
|
||||
from services.lstm_predictor import LSTMPredictor
|
||||
predictor = LSTMPredictor(db_path='backend/instance/app.sqlite')
|
||||
|
||||
if hasattr(predictor, 'extract_features'):
|
||||
features = predictor.extract_features(1)
|
||||
assert features is not None
|
||||
except Exception as e:
|
||||
pytest.skip(f"特征提取测试失败: {e}")
|
||||
@ -1,103 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
统计服务单元测试
|
||||
"""
|
||||
import pytest
|
||||
from services.stats_service import StatsService
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.stats
|
||||
class TestStatsService:
|
||||
"""统计服务测试"""
|
||||
|
||||
def test_get_user_stats_basic(self, stats_service):
|
||||
"""测试获取用户基本统计信息"""
|
||||
stats = stats_service.get_user_stats(1)
|
||||
|
||||
assert stats is not None
|
||||
assert 'total_submissions' in stats
|
||||
assert 'accepted_submissions' in stats
|
||||
assert 'acceptance_rate' in stats
|
||||
assert 'rank' in stats
|
||||
assert 'language_distribution' in stats
|
||||
|
||||
def test_get_user_stats_submissions_count(self, stats_service):
|
||||
"""测试提交数量统计"""
|
||||
stats = stats_service.get_user_stats(1)
|
||||
|
||||
assert stats['total_submissions'] == 3
|
||||
assert stats['accepted_submissions'] == 2
|
||||
|
||||
def test_get_user_stats_acceptance_rate(self, stats_service):
|
||||
"""测试通过率计算"""
|
||||
stats = stats_service.get_user_stats(1)
|
||||
|
||||
expected_rate = (2 / 3) * 100
|
||||
assert abs(stats['acceptance_rate'] - expected_rate) < 0.1
|
||||
|
||||
def test_get_user_stats_language_distribution(self, stats_service):
|
||||
"""测试语言分布统计"""
|
||||
stats = stats_service.get_user_stats(1)
|
||||
|
||||
lang_dist = stats['language_distribution']
|
||||
assert len(lang_dist) >= 1
|
||||
|
||||
# 检查语言分布格式
|
||||
for lang_stat in lang_dist:
|
||||
assert 'language' in lang_stat
|
||||
assert 'total_submissions' in lang_stat
|
||||
assert 'accepted_submissions' in lang_stat
|
||||
assert 'acceptance_rate' in lang_stat
|
||||
|
||||
def test_get_user_stats_nonexistent_user(self, stats_service):
|
||||
"""测试获取不存在用户的统计"""
|
||||
stats = stats_service.get_user_stats(999)
|
||||
|
||||
assert stats is not None
|
||||
assert stats['total_submissions'] == 0
|
||||
assert stats['accepted_submissions'] == 0
|
||||
|
||||
def test_get_user_stats_rank(self, stats_service):
|
||||
"""测试用户排名"""
|
||||
stats1 = stats_service.get_user_stats(1)
|
||||
stats2 = stats_service.get_user_stats(2)
|
||||
|
||||
assert 'rank' in stats1
|
||||
assert 'rank' in stats2
|
||||
assert stats1['rank'] >= 1
|
||||
assert stats2['rank'] >= 1
|
||||
|
||||
def test_get_language_stats(self, stats_service):
|
||||
"""测试获取语言统计"""
|
||||
try:
|
||||
lang_stats = stats_service.get_language_stats()
|
||||
assert lang_stats is not None
|
||||
assert isinstance(lang_stats, list)
|
||||
except AttributeError:
|
||||
# 如果方法不存在，跳过测试
|
||||
pytest.skip("get_language_stats方法不存在")
|
||||
|
||||
def test_cache_functionality(self, stats_service):
|
||||
"""测试缓存功能"""
|
||||
# 第一次调用
|
||||
stats1 = stats_service.get_user_stats(1)
|
||||
# 第二次调用（应该从缓存获取）
|
||||
stats2 = stats_service.get_user_stats(1)
|
||||
|
||||
assert stats1 == stats2
|
||||
|
||||
def test_get_user_stats_with_details(self, stats_service):
|
||||
"""测试获取详细统计信息"""
|
||||
stats = stats_service.get_user_stats(1)
|
||||
|
||||
# 检查是否包含详细统计
|
||||
if 'difficulty_stats' in stats:
|
||||
assert isinstance(stats['difficulty_stats'], (list, dict))
|
||||
|
||||
if 'time_trends' in stats:
|
||||
assert isinstance(stats['time_trends'], (list, dict))
|
||||
|
||||
if 'topic_stats' in stats:
|
||||
assert isinstance(stats['topic_stats'], (list, dict))
|
||||
@ -1,39 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
建议服务单元测试
|
||||
"""
|
||||
import pytest
|
||||
from services.suggestion_service import SuggestionService
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestSuggestionService:
|
||||
"""建议服务测试"""
|
||||
|
||||
def test_get_suggestions_structure(self, suggestion_service):
|
||||
"""测试建议结构"""
|
||||
try:
|
||||
suggestions = suggestion_service.get_suggestions(1)
|
||||
assert suggestions is not None
|
||||
assert isinstance(suggestions, (list, dict))
|
||||
except Exception as e:
|
||||
pytest.skip(f"get_suggestions方法异常: {e}")
|
||||
|
||||
def test_get_problem_recommendations(self, suggestion_service):
|
||||
"""测试问题推荐"""
|
||||
try:
|
||||
if hasattr(suggestion_service, 'get_problem_recommendations'):
|
||||
recommendations = suggestion_service.get_problem_recommendations(1)
|
||||
assert isinstance(recommendations, list)
|
||||
except Exception:
|
||||
pytest.skip("get_problem_recommendations方法不可用")
|
||||
|
||||
def test_get_learning_path(self, suggestion_service):
|
||||
"""测试学习路径"""
|
||||
try:
|
||||
if hasattr(suggestion_service, 'get_learning_path'):
|
||||
path = suggestion_service.get_learning_path(1)
|
||||
assert path is not None
|
||||
except Exception:
|
||||
pytest.skip("get_learning_path方法不可用")
|
||||
@ -1,131 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
SonarQube覆盖率验证脚本
|
||||
验证coverage.xml是否正确生成并可被SonarQube识别
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
|
||||
def check_coverage_xml():
|
||||
"""检查coverage.xml文件"""
|
||||
print("=" * 70)
|
||||
print("SonarQube 覆盖率报告验证")
|
||||
print("=" * 70)
|
||||
|
||||
# 检查文件是否存在
|
||||
if not os.path.exists('coverage.xml'):
|
||||
print("❌ 错误: coverage.xml 文件不存在")
|
||||
print("\n请先运行测试生成覆盖率报告:")
|
||||
print(" python run_tests.py")
|
||||
print(" 或")
|
||||
print(" pytest tests/ --cov=. --cov-report=xml:coverage.xml")
|
||||
return False
|
||||
|
||||
print("✓ coverage.xml 文件存在")
|
||||
|
||||
# 检查文件大小
|
||||
file_size = os.path.getsize('coverage.xml')
|
||||
print(f"✓ 文件大小: {file_size:,} bytes")
|
||||
|
||||
if file_size < 100:
|
||||
print("❌ 警告: 文件太小，可能没有包含有效数据")
|
||||
return False
|
||||
|
||||
# 解析XML文件
|
||||
try:
|
||||
tree = ET.parse('coverage.xml')
|
||||
root = tree.getroot()
|
||||
print("✓ XML格式有效")
|
||||
|
||||
# 获取覆盖率信息
|
||||
line_rate = float(root.get('line-rate', 0))
|
||||
lines_valid = int(root.get('lines-valid', 0))
|
||||
lines_covered = int(root.get('lines-covered', 0))
|
||||
|
||||
coverage_percent = line_rate * 100
|
||||
|
||||
print(f"\n📊 覆盖率统计:")
|
||||
print(f" 总代码行数: {lines_valid:,}")
|
||||
print(f" 已覆盖行数: {lines_covered:,}")
|
||||
print(f" 覆盖率: {coverage_percent:.2f}%")
|
||||
|
||||
# 检查是否有包信息
|
||||
packages = root.findall('.//package')
|
||||
print(f"\n📦 包数量: {len(packages)}")
|
||||
|
||||
# 检查是否有类信息
|
||||
classes = root.findall('.//class')
|
||||
print(f"📄 文件数量: {len(classes)}")
|
||||
|
||||
if len(classes) > 0:
|
||||
print("\n前5个已分析的文件:")
|
||||
for cls in classes[:5]:
|
||||
filename = cls.get('filename', 'unknown')
|
||||
cls_line_rate = float(cls.get('line-rate', 0))
|
||||
print(f" - {filename}: {cls_line_rate*100:.2f}%")
|
||||
|
||||
print("\n" + "=" * 70)
|
||||
print("✅ coverage.xml 文件有效，可以被 SonarQube 识别")
|
||||
print("=" * 70)
|
||||
|
||||
print("\n下一步:")
|
||||
print("1. 确保 SonarQube 服务正在运行")
|
||||
print("2. 运行 sonar-scanner 进行代码分析")
|
||||
print("3. 访问 http://localhost:9000 查看结果")
|
||||
|
||||
return True
|
||||
|
||||
except ET.ParseError as e:
|
||||
print(f"❌ XML解析错误: {e}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"❌ 未知错误: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def check_sonar_config():
|
||||
"""检查SonarQube配置"""
|
||||
print("\n" + "=" * 70)
|
||||
print("检查 SonarQube 配置")
|
||||
print("=" * 70)
|
||||
|
||||
if not os.path.exists('sonar-project.properties'):
|
||||
print("❌ sonar-project.properties 文件不存在")
|
||||
return False
|
||||
|
||||
print("✓ sonar-project.properties 文件存在")
|
||||
|
||||
with open('sonar-project.properties', 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
|
||||
# 检查关键配置
|
||||
checks = {
|
||||
'sonar.python.coverage.reportPaths': 'coverage.xml',
|
||||
'sonar.sources': ['app.py', 'services', 'config.py'],
|
||||
'sonar.tests': 'tests'
|
||||
}
|
||||
|
||||
all_ok = True
|
||||
for key, expected in checks.items():
|
||||
if key in content:
|
||||
print(f"✓ {key} 已配置")
|
||||
else:
|
||||
print(f"❌ {key} 未配置")
|
||||
all_ok = False
|
||||
|
||||
return all_ok
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
success = check_coverage_xml()
|
||||
config_ok = check_sonar_config()
|
||||
|
||||
if success and config_ok:
|
||||
print("\n🎉 所有检查通过! 可以运行 SonarQube 分析")
|
||||
sys.exit(0)
|
||||
else:
|
||||
print("\n⚠️ 存在问题，请修复后再运行 SonarQube 分析")
|
||||
sys.exit(1)
|
||||
Loading…
Reference in new issue