reflector/server/tests/test_transcripts.py

import pytest
from unittest.mock import patch
from contextlib import asynccontextmanager
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_transcript_create():
    from reflector.app import app

    async with AsyncClient(app=app, base_url="http://test/v1") as ac:
        response = await ac.post("/transcripts", json={"name": "test"})
        assert response.status_code == 200
        assert response.json()["name"] == "test"
        assert response.json()["status"] == "idle"
        assert response.json()["locked"] is False
        assert response.json()["id"] is not None
        assert response.json()["created_at"] is not None

        # ensure some fields are not returned
        assert "topics" not in response.json()
        assert "events" not in response.json()


@pytest.mark.asyncio
async def test_transcript_get_update_name():
    from reflector.app import app

    async with AsyncClient(app=app, base_url="http://test/v1") as ac:
        response = await ac.post("/transcripts", json={"name": "test"})
        assert response.status_code == 200
        assert response.json()["name"] == "test"

        tid = response.json()["id"]

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["name"] == "test"

        response = await ac.patch(f"/transcripts/{tid}", json={"name": "test2"})
        assert response.status_code == 200
        assert response.json()["name"] == "test2"

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["name"] == "test2"


@pytest.mark.asyncio
async def test_transcript_get_update_locked():
    from reflector.app import app

    async with AsyncClient(app=app, base_url="http://test/v1") as ac:
        response = await ac.post("/transcripts", json={"name": "test"})
        assert response.status_code == 200
        assert response.json()["locked"] is False

        tid = response.json()["id"]

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["locked"] is False

        response = await ac.patch(f"/transcripts/{tid}", json={"locked": True})
        assert response.status_code == 200
        assert response.json()["locked"] is True

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["locked"] is True


@pytest.mark.asyncio
async def test_transcript_get_update_summary():
    from reflector.app import app

    async with AsyncClient(app=app, base_url="http://test/v1") as ac:
        response = await ac.post("/transcripts", json={"name": "test"})
        assert response.status_code == 200
        assert response.json()["long_summary"] is None
        assert response.json()["short_summary"] is None

        tid = response.json()["id"]

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["long_summary"] is None
        assert response.json()["short_summary"] is None

        response = await ac.patch(
            f"/transcripts/{tid}",
            json={"long_summary": "test_long", "short_summary": "test_short"},
        )
        assert response.status_code == 200
        assert response.json()["long_summary"] == "test_long"
        assert response.json()["short_summary"] == "test_short"

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["long_summary"] == "test_long"
        assert response.json()["short_summary"] == "test_short"


@pytest.mark.asyncio
async def test_transcript_get_update_title():
    from reflector.app import app

    async with AsyncClient(app=app, base_url="http://test/v1") as ac:
        response = await ac.post("/transcripts", json={"name": "test"})
        assert response.status_code == 200
        assert response.json()["title"] is None

        tid = response.json()["id"]

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["title"] is None

        response = await ac.patch(f"/transcripts/{tid}", json={"title": "test_title"})
        assert response.status_code == 200
        assert response.json()["title"] == "test_title"

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["title"] == "test_title"


@pytest.mark.asyncio
async def test_transcripts_list_anonymous():
    # XXX this test is a bit fragile, as it depends on the storage which
    #     is shared between tests
    from reflector.app import app
    from reflector.settings import settings

    async with AsyncClient(app=app, base_url="http://test/v1") as ac:
        response = await ac.get("/transcripts")
        assert response.status_code == 401

    # if public mode, it should be allowed
    try:
        settings.PUBLIC_MODE = True
        async with AsyncClient(app=app, base_url="http://test/v1") as ac:
            response = await ac.get("/transcripts")
            assert response.status_code == 200
    finally:
        settings.PUBLIC_MODE = False


@asynccontextmanager
async def authenticated_client_ctx():
    from reflector.app import app
    from reflector.auth import current_user, current_user_optional

    app.dependency_overrides[current_user] = lambda: {
        "sub": "randomuserid",
        "email": "test@mail.com",
    }
    app.dependency_overrides[current_user_optional] = lambda: {
        "sub": "randomuserid",
        "email": "test@mail.com",
    }
    yield
    del app.dependency_overrides[current_user]
    del app.dependency_overrides[current_user_optional]


@asynccontextmanager
async def authenticated_client2_ctx():
    from reflector.app import app
    from reflector.auth import current_user, current_user_optional

    app.dependency_overrides[current_user] = lambda: {
        "sub": "randomuserid2",
        "email": "test@mail.com",
    }
    app.dependency_overrides[current_user_optional] = lambda: {
        "sub": "randomuserid2",
        "email": "test@mail.com",
    }
    yield
    del app.dependency_overrides[current_user]
    del app.dependency_overrides[current_user_optional]


@pytest.fixture
@pytest.mark.asyncio
async def authenticated_client():
    async with authenticated_client_ctx():
        yield


@pytest.fixture
@pytest.mark.asyncio
async def authenticated_client2():
    async with authenticated_client2_ctx():
        yield


@pytest.mark.asyncio
async def test_transcripts_list_authenticated(authenticated_client):
    # XXX this test is a bit fragile, as it depends on the storage which
    #     is shared between tests
    from reflector.app import app

    async with AsyncClient(app=app, base_url="http://test/v1") as ac:
        response = await ac.post("/transcripts", json={"name": "testxx1"})
        assert response.status_code == 200
        assert response.json()["name"] == "testxx1"

        response = await ac.post("/transcripts", json={"name": "testxx2"})
        assert response.status_code == 200
        assert response.json()["name"] == "testxx2"

        response = await ac.get("/transcripts")
        assert response.status_code == 200
        assert len(response.json()["items"]) >= 2
        names = [t["name"] for t in response.json()["items"]]
        assert "testxx1" in names
        assert "testxx2" in names


@pytest.mark.asyncio
async def test_transcript_delete():
    from reflector.app import app

    async with AsyncClient(app=app, base_url="http://test/v1") as ac:
        response = await ac.post("/transcripts", json={"name": "testdel1"})
        assert response.status_code == 200
        assert response.json()["name"] == "testdel1"

        tid = response.json()["id"]
        response = await ac.delete(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["status"] == "ok"

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 404


@pytest.mark.asyncio
async def test_transcript_mark_reviewed():
    from reflector.app import app

    async with AsyncClient(app=app, base_url="http://test/v1") as ac:
        response = await ac.post("/transcripts", json={"name": "test"})
        assert response.status_code == 200
        assert response.json()["name"] == "test"
        assert response.json()["reviewed"] is False

        tid = response.json()["id"]

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["name"] == "test"
        assert response.json()["reviewed"] is False

        response = await ac.patch(f"/transcripts/{tid}", json={"reviewed": True})
        assert response.status_code == 200
        assert response.json()["reviewed"] is True

        response = await ac.get(f"/transcripts/{tid}")
        assert response.status_code == 200
        assert response.json()["reviewed"] is True


@asynccontextmanager
async def patch_migrate_user():
    with patch(
        "reflector.db.migrate_user.users_to_migrate",
        [["test@mail.com", "randomuserid", None]],
    ):
        yield


@pytest.mark.asyncio
async def test_transcripts_list_authenticated_migration():
    # XXX this test is a bit fragile, as it depends on the storage which
    #     is shared between tests
    from reflector.app import app

    testx1 = "testmigration1"
    testx2 = "testmigration2"

    async with patch_migrate_user(), AsyncClient(
        app=app, base_url="http://test/v1"
    ) as ac:
        # first ensure client 2 does not have any transcripts related to this test
        async with authenticated_client2_ctx():
            response = await ac.get("/transcripts")
            assert response.status_code == 200
            # assert len(response.json()["items"]) == 0
            names = [t["name"] for t in response.json()["items"]]
            assert testx1 not in names
            assert testx2 not in names

        # create 2 transcripts with client 1
        async with authenticated_client_ctx():
            response = await ac.post("/transcripts", json={"name": testx1})
            assert response.status_code == 200
            assert response.json()["name"] == testx1

            response = await ac.post("/transcripts", json={"name": testx2})
            assert response.status_code == 200
            assert response.json()["name"] == testx2

            response = await ac.get("/transcripts")
            assert response.status_code == 200
            assert len(response.json()["items"]) >= 2
            names = [t["name"] for t in response.json()["items"]]
            assert testx1 in names
            assert testx2 in names

        # now going back to client 2, migration should happen
        async with authenticated_client2_ctx():
            response = await ac.get("/transcripts")
            assert response.status_code == 200
            names = [t["name"] for t in response.json()["items"]]
            assert testx1 in names
            assert testx2 in names

        # and client 1 should have nothing now
        async with authenticated_client_ctx():
            response = await ac.get("/transcripts")
            assert response.status_code == 200
            names = [t["name"] for t in response.json()["items"]]
            assert testx1 not in names
            assert testx2 not in names