import json from dataclasses import replace from fastapi.testclient import TestClient from sentence_api.main import create_app from sentence_api.config import Settings from sentence_api.oss import VolcanoOSSUploader from sentence_api.models import SentenceBoundary, SentenceBoundaryDocument from sentence_api.processing import VideoProcessor from sentence_api.repository import VideoRepository from sentence_api.store import BoundaryStore VIDEO_HASH = "a" * 64 def make_client(tmp_path, admin_api_key=""): index_path = tmp_path / "boundaries.json" index_path.write_text( json.dumps({ "videos": { VIDEO_HASH: { "duration_ms": 9000, "algorithm_version": "test-v1", "sentences": [ {"index": 0, "start_ms": 0, "end_ms": 4000, "text": "One."}, {"index": 1, "start_ms": 4000, "end_ms": 9000, "text": "Two."}, ], } } }), encoding="utf-8", ) settings = replace( Settings.from_env(), data_dir=tmp_path / "data", legacy_boundaries_file=index_path, admin_api_key=admin_api_key, ) settings.ensure_directories() client = TestClient( create_app( BoundaryStore(index_path), settings=settings, repository=VideoRepository(settings.database_path), ) ) response = client.post( "/api/v1/auth/register", json={"username": "tester", "password": "password-123"}, ) assert response.status_code == 201, response.text client.headers.update({"Authorization": f"Bearer {response.json()['token']}"}) return client def test_lookup_returns_boundaries(tmp_path): client = make_client(tmp_path) response = client.get(f"/api/v1/videos/{VIDEO_HASH}/sentence-boundaries") assert response.status_code == 200 assert response.json()["sentences"][1]["start_ms"] == 4000 def test_unknown_hash_returns_not_found(tmp_path): client = make_client(tmp_path) response = client.get(f"/api/v1/videos/{'b' * 64}/sentence-boundaries") assert response.status_code == 404 assert response.json()["detail"]["code"] == "SENTENCE_BOUNDARIES_NOT_FOUND" def test_invalid_hash_is_rejected(tmp_path): client = make_client(tmp_path) response = client.get("/api/v1/videos/not-a-sha256/sentence-boundaries") assert response.status_code == 422 def test_adhoc_audio_boundaries_do_not_persist_audio(tmp_path, monkeypatch): client = make_client(tmp_path) upload = tmp_path / "audio.wav" upload.write_bytes(b"not-a-real-audio") import sentence_api.processing as processing_module monkeypatch.setattr(processing_module, "detect_sentence_boundaries", lambda path: [0.0, 1.0]) monkeypatch.setattr(processing_module, "media_duration_ms", lambda path: 2000) response = client.post( "/api/v1/audio/sentence-boundaries", files={"file": ("audio.wav", upload.read_bytes(), "audio/wav")}, ) assert response.status_code == 200, response.text payload = response.json() assert payload["algorithm_version"] == "silence-rms-v1" assert payload["duration_ms"] == 2000 assert [(item["start_ms"], item["end_ms"]) for item in payload["sentences"]] == [(0, 1000), (1000, 2000)] assert not (tmp_path / "data" / "work").exists() or not any((tmp_path / "data" / "work").iterdir()) assert not (tmp_path / "data" / "v").exists() or not any((tmp_path / "data" / "v").iterdir()) def test_adhoc_audio_boundaries_require_login(tmp_path): client = make_client(tmp_path) client.headers.clear() response = client.post( "/api/v1/audio/sentence-boundaries", files={"file": ("audio.wav", b"audio", "audio/wav")}, ) assert response.status_code == 401 def test_admin_can_reset_sso_user_password(tmp_path): client = make_client(tmp_path, admin_api_key="test-admin-key") repository = client.app.state.video_repository created = repository.create_sso_user( username="bobcoc", nickname="Bob", sso_provider="d1kt", ) assert created is not None assert repository.authenticate_user("bobcoc", "anything-123") is None response = client.put( "/api/v1/admin/users/password", json={"username": "bobcoc", "password": "new-password-123"}, headers={"X-Admin-Key": "test-admin-key"}, ) assert response.status_code == 204 assert repository.authenticate_user("bobcoc", "new-password-123") is not None def test_sentence_explanation_requires_login(tmp_path): index_path = tmp_path / "boundaries.json" index_path.write_text("{}", encoding="utf-8") settings = replace( Settings.from_env(), data_dir=tmp_path / "data", legacy_boundaries_file=index_path, ) settings.ensure_directories() client = TestClient( create_app( BoundaryStore(index_path), settings=settings, repository=VideoRepository(settings.database_path), ) ) response = client.post( "/api/v1/sentences/explain", json={"sentence_text": "Hello."}, ) assert response.status_code == 401 def test_sentence_explanation_uses_server_side_ai_client(tmp_path, monkeypatch): client = make_client(tmp_path) requested = [] def explain(text, **kwargs): requested.append((text, kwargs)) return "这句话表示打招呼。" class StubAITutor: pass stub = StubAITutor() stub.explain = explain monkeypatch.setattr(client.app.state, "ai_tutor", stub) response = client.post( "/api/v1/sentences/explain", json={ "sentence_text": "Hello.", "course_title": "Greetings", "sentence_index": 0, "language": "en", }, ) assert response.status_code == 200 assert response.json() == {"explanation": "这句话表示打招呼。"} assert requested == [ ( "Hello.", { "course_title": "Greetings", "sentence_index": 0, "language": "en", }, ) ] def test_oss_video_url_is_used_for_catalog_and_content(tmp_path): index_path = tmp_path / "boundaries.json" index_path.write_text("{}", encoding="utf-8") settings = replace( Settings.from_env(), data_dir=tmp_path / "data", legacy_boundaries_file=index_path, ) settings.ensure_directories() repository = VideoRepository(settings.database_path) repository.upsert_upload( video_hash=VIDEO_HASH, title="OSS lesson", filename="lesson.mp4", stored_filename=f"{VIDEO_HASH}.mp4", content_type="video/mp4", size_bytes=1, language=None, ) remote_url = f"https://course-media.example.com/videos/{VIDEO_HASH}.mp4" repository.mark_oss_uploaded(VIDEO_HASH, remote_url) client = TestClient( create_app( BoundaryStore(index_path), settings=settings, repository=repository, ) ) catalog = client.get("/api/v1/videos") stream = client.get(f"/api/v1/videos/{VIDEO_HASH}/content", follow_redirects=False) assert catalog.status_code == 200 assert catalog.json()["videos"][0]["stream_url"] == remote_url assert stream.status_code == 307 assert stream.headers["location"] == remote_url def test_volcano_oss_object_key_and_public_url(tmp_path): settings = replace( Settings.from_env(), volcano_oss_access_key="ak", volcano_oss_secret_key="sk", volcano_oss_bucket="course-media", volcano_oss_endpoint="tos-cn-beijing.volces.com", volcano_oss_region="cn-beijing", volcano_oss_key_prefix="lessons/videos/", ) uploader = VolcanoOSSUploader(settings) key = uploader.object_key(VIDEO_HASH, f"{VIDEO_HASH}.mp4") assert settings.volcano_oss_enabled is True assert key == f"lessons/videos/{VIDEO_HASH}.mp4" assert uploader.public_url(key) == ( f"https://course-media.tos-cn-beijing.volces.com/" f"lessons/videos/{VIDEO_HASH}.mp4" ) def test_processor_marks_ready_only_after_oss_upload(tmp_path, monkeypatch): settings = replace(Settings.from_env(), data_dir=tmp_path / "data") settings.ensure_directories() media_path = settings.videos_dir / f"{VIDEO_HASH}.mp4" media_path.write_bytes(b"video") repository = VideoRepository(settings.database_path) repository.upsert_upload( video_hash=VIDEO_HASH, title="OSS lesson", filename="lesson.mp4", stored_filename=f"{VIDEO_HASH}.mp4", content_type="video/mp4", size_bytes=5, language=None, ) document = SentenceBoundaryDocument( video_hash=VIDEO_HASH, duration_ms=1000, algorithm_version="test", sentences=[SentenceBoundary(index=0, start_ms=0, end_ms=1000)], ) uploads = [] class EnabledUploader: enabled = True def upload(self, **kwargs): uploads.append(kwargs) return "https://media.example.com/lesson.mp4" class UnavailableTranscriber: available = False monkeypatch.setattr( "sentence_api.processing.make_entry", lambda *args, **kwargs: ( { "duration_ms": 1000, "sentences": [{"index": 0, "start_ms": 0, "end_ms": 1000}], }, {}, ), ) processor = VideoProcessor( settings, repository, UnavailableTranscriber(), EnabledUploader(), ) processor.process(VIDEO_HASH) video = repository.get_video(VIDEO_HASH) assert uploads[0]["stored_filename"] == f"{VIDEO_HASH}.mp4" assert video["status"] == "ready" assert video["remote_url"] == "https://media.example.com/lesson.mp4" def test_admin_can_split_sentence_and_reindexes_following_sentences(tmp_path): settings = replace(Settings.from_env(), data_dir=tmp_path / "data") settings.ensure_directories() repository = VideoRepository(settings.database_path) repository.upsert_upload( video_hash=VIDEO_HASH, title="Manual split lesson", filename="lesson.mp4", stored_filename=f"{VIDEO_HASH}.mp4", content_type="video/mp4", size_bytes=5, language="en", ) repository.save_processing_result( SentenceBoundaryDocument( video_hash=VIDEO_HASH, duration_ms=12000, algorithm_version="test", sentences=[ SentenceBoundary( index=0, start_ms=0, end_ms=2000, text="Before.", language="en", reference_speech_duration_ms=1800, ), SentenceBoundary( index=1, start_ms=2000, end_ms=9000, text="First part second part", language="en", reference_speech_duration_ms=6600, ), SentenceBoundary( index=2, start_ms=9000, end_ms=12000, text="Next sentence.", language="en", reference_speech_duration_ms=2800, ), ], ), None, ) class UnavailableTranscriber: available = False client = TestClient( create_app( BoundaryStore(tmp_path / "boundaries.json"), settings=settings, repository=repository, transcriber=UnavailableTranscriber(), ) ) response = client.put( f"/api/v1/admin/videos/{VIDEO_HASH}/sentences/1/split", json={"text": "First part\nsecond part", "language": "en"}, ) assert response.status_code == 200 replacements = response.json() assert [item["text"] for item in replacements] == ["First part", "second part"] assert replacements[0]["start_ms"] == 2000 assert replacements[0]["end_ms"] == replacements[1]["start_ms"] assert replacements[1]["end_ms"] == 9000 document = repository.get_document(VIDEO_HASH) assert [item.text for item in document.sentences] == [ "Before.", "First part", "second part", "Next sentence.", ] assert [item.index for item in document.sentences] == [0, 1, 2, 3] def test_admin_can_merge_adjacent_sentences_and_reindexes_following_sentences(tmp_path): settings = replace(Settings.from_env(), data_dir=tmp_path / "data") settings.ensure_directories() repository = VideoRepository(settings.database_path) repository.upsert_upload( video_hash=VIDEO_HASH, title="Manual merge lesson", filename="lesson.mp4", stored_filename=f"{VIDEO_HASH}.mp4", content_type="video/mp4", size_bytes=5, language="en", ) repository.save_processing_result( SentenceBoundaryDocument( video_hash=VIDEO_HASH, duration_ms=12000, algorithm_version="test", sentences=[ SentenceBoundary( index=0, start_ms=0, end_ms=2000, text="Before.", language="en", reference_speech_duration_ms=1800, ), SentenceBoundary( index=1, start_ms=2000, end_ms=5000, text="First part", language="en", reference_speech_duration_ms=2800, ), SentenceBoundary( index=2, start_ms=5000, end_ms=9000, text="second part", language="en", reference_speech_duration_ms=3600, ), SentenceBoundary( index=3, start_ms=9000, end_ms=12000, text="Next sentence.", language="en", reference_speech_duration_ms=2800, ), ], ), None, ) class UnavailableTranscriber: available = False client = TestClient( create_app( BoundaryStore(tmp_path / "boundaries.json"), settings=settings, repository=repository, transcriber=UnavailableTranscriber(), ) ) response = client.put( f"/api/v1/admin/videos/{VIDEO_HASH}/sentences/1/merge", ) assert response.status_code == 200 merged = response.json() assert merged["index"] == 1 assert merged["text"] == "First part second part" assert merged["start_ms"] == 2000 assert merged["end_ms"] == 9000 assert merged["reference_speech_duration_ms"] == 6400 document = repository.get_document(VIDEO_HASH) assert [item.text for item in document.sentences] == [ "Before.", "First part second part", "Next sentence.", ] assert [item.index for item in document.sentences] == [0, 1, 2] assert [item.start_ms for item in document.sentences] == [0, 2000, 9000] assert [item.end_ms for item in document.sentences] == [2000, 9000, 12000] last_response = client.put( f"/api/v1/admin/videos/{VIDEO_HASH}/sentences/2/merge", ) assert last_response.status_code == 400 assert last_response.json()["detail"] == "No next sentence to merge with." def test_admin_boundary_adjustment_updates_linked_next_start(tmp_path): settings = replace(Settings.from_env(), data_dir=tmp_path / "data") settings.ensure_directories() repository = VideoRepository(settings.database_path) repository.upsert_upload( video_hash=VIDEO_HASH, title="Boundary lesson", filename="lesson.mp4", stored_filename=f"{VIDEO_HASH}.mp4", content_type="video/mp4", size_bytes=5, language="en", ) repository.save_processing_result( SentenceBoundaryDocument( video_hash=VIDEO_HASH, duration_ms=9000, algorithm_version="test", sentences=[ SentenceBoundary( index=0, start_ms=0, end_ms=4000, text="First.", language="en", reference_speech_duration_ms=3600, ), SentenceBoundary( index=1, start_ms=4000, end_ms=9000, text="Second.", language="en", reference_speech_duration_ms=4600, ), ], ), None, ) class UnavailableTranscriber: available = False client = TestClient( create_app( BoundaryStore(tmp_path / "boundaries.json"), settings=settings, repository=repository, transcriber=UnavailableTranscriber(), ) ) response = client.put( f"/api/v1/admin/videos/{VIDEO_HASH}/sentences/0/boundary", json={"delta_ms": 250}, ) assert response.status_code == 200 updated = response.json() assert [item["index"] for item in updated] == [0, 1] assert updated[0]["end_ms"] == 4250 assert updated[1]["start_ms"] == 4250 assert updated[0]["reference_speech_duration_ms"] == 3850 assert updated[1]["reference_speech_duration_ms"] == 4350 document = repository.get_document(VIDEO_HASH) assert [item.end_ms for item in document.sentences] == [4250, 9000] assert [item.start_ms for item in document.sentences] == [0, 4250]