foxhole/tests/test_outbox.py

#!/usr/bin/env python3
import pytest
import fastapi
import respx
import httpx
from uuid import uuid4
from tests import factories
from tests.utils import build_remote_actor

from unittest import mock
from app.main import app
from app.utils import precheck
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app import activitypub as ap, httpsig
from app.config import AP_CONTENT_TYPE
from app import models
from sqlalchemy import select


@pytest.mark.asyncio
async def test_outbox_send_follow_request(
    db: Session,
    async_db_session,
    client: TestClient,
    respx_mock: respx.MockRouter,
) -> None:
    # build test actor
    ra = build_remote_actor()
    remote_ap_id = ra.ap_id # type: ignore

    # mock request
    respx_mock.get(remote_ap_id).mock(
        return_value=httpx.Response(200,json=ra.ap_actor))
    respx_mock.post(remote_ap_id + "/inbox").mock(
        return_value=httpx.Response(202))

    from app.boxes import send_follow

    async with async_db_session as db_session: #type: ignore
        await send_follow(db_session, remote_ap_id)

    # And the Follow activity was created in the outbox
    outbox_object = db.execute(select(models.OutboxObject)).scalar_one()
    assert outbox_object.ap_type == "Follow"
    assert outbox_object.activity_object_ap_id == remote_ap_id

    response = client.post(
        "/inbox",
        headers={"Content-Type": AP_CONTENT_TYPE},
        json={
            "@context": ap.AS_CTX,
            "type": "Accept",
            "id": "https://accept-test",
            "actor": remote_ap_id,
            "object": outbox_object.ap_id,
        },
    )

    assert response.status_code == 202

    # follower was saved in follower table
    following_actor = db.execute(select(models.Following)).scalar_one()
    assert following_actor.ap_actor_id == remote_ap_id
    assert following_actor.outbox_object == outbox_object

    inbox_object = db.execute(select(models.InboxObject)).scalar_one()
    assert inbox_object.ap_type == "Accept"


@pytest.mark.asyncio
async def test_outbox_send_follow_request_nest_accept(
    db: Session,
    async_db_session,
    client: TestClient,
    respx_mock: respx.MockRouter,
) -> None:
    # build test actor
    ra = build_remote_actor()
    remote_ap_id = ra.ap_id # type: ignore

    # mock request
    respx_mock.get(remote_ap_id).mock(
        return_value=httpx.Response(200,json=ra.ap_actor))
    respx_mock.post(remote_ap_id + "/inbox").mock(
        return_value=httpx.Response(202))

    from app.boxes import send_follow

    async with async_db_session as db_session: #type: ignore
        await send_follow(db_session, remote_ap_id)

    # And the Follow activity was created in the outbox
    outbox_object = db.execute(select(models.OutboxObject)).scalar_one()
    assert outbox_object.ap_type == "Follow"
    assert outbox_object.activity_object_ap_id == remote_ap_id

    response = client.post(
        "/inbox",
        headers={"Content-Type": AP_CONTENT_TYPE},
        json={
            "@context": ap.AS_CTX,
            "type": "Accept",
            "id": "https://accept-test",
            "actor": remote_ap_id,
            "object": {
                "id": outbox_object.ap_id,
                "type": "Accept",
                "actor": ap.ME["id"],
                "object": remote_ap_id,
            },
        },
    )

    assert response.status_code == 202

    # follower was saved in follower table
    following_actor = db.execute(select(models.Following)).scalar_one()
    assert following_actor.ap_actor_id == remote_ap_id


@pytest.mark.asyncio
async def test_outbox_send_create_activity(
    db: Session,
    async_db_session,
    respx_mock: respx.MockRouter,
    mocker,
) -> None:
    # build test actor
    ra = build_remote_actor()
    remote_ap_id = ra.ap_id

    # mock request
    respx_mock.get(remote_ap_id).mock(
        return_value=httpx.Response(200,json=ra.ap_actor))
    respx_mock.post(remote_ap_id + "/inbox").mock(
        return_value=httpx.Response(202))

    from app.boxes import _send_create
    from app.activitypub import VisibilityEnum
    from app.orgpython import to_html

    mocker.patch("app.ldsig.generate_signature", return_value=True)
    async with async_db_session as db_session:
        content = "*Blod Text* =code Text= \n"
        content = to_html(content)

        await _send_create(
            db_session,
            "Note",
            content,
            VisibilityEnum.PUBLIC
        )

    # And the Follow activity was created in the outbox
    outbox_object = db.execute(select(models.OutboxObject)).scalar_one()
    assert "Note" == outbox_object.ap_type
    assert outbox_object.ap_object["content"] == content


@pytest.mark.asyncio
async def test_outbox_send_unlisted_note(
    db: Session,
    async_db_session,
    client: TestClient,
    respx_mock: respx.MockRouter,
) -> None:
    # build test actor
    ra = build_remote_actor()
    remote_ap_id = ra.ap_id # type: ignore

    # mock request
    respx_mock.get(remote_ap_id).mock(
        return_value=httpx.Response(200,json=ra.ap_actor))
    respx_mock.post(remote_ap_id + "/inbox").mock(
        return_value=httpx.Response(202))

    from app.activitypub import VisibilityEnum


    response = client.post(
        "/outbox",
        headers={"Authorization": "Basic test-token"},
        content='{"visibility": "unlisted","content": "note content"}'
    )

    assert response.status_code == 200
    # And the Follow activity was created in the outbox
    outbox_object = db.execute(select(models.OutboxObject)).scalar_one()
    assert outbox_object.ap_type == "Note"
    assert outbox_object.visibility == VisibilityEnum.UNLISTED