ap-test/features/steps/object.py

# SPDX-FileCopyrightText: 2023 Aravinth Manivannan <realaravinth@batsense.net>
#
# SPDX-License-Identifier: AGPL-3.0-or-later

import os
from urllib.parse import urlparse, urlunparse

import requests
from behave import *

from ftest_common.env import FTEST_USER
from ftest_common.logger import logger
from ftest_common.webfinger import get_webfinger
from ftest_common.obj import get_ap_obj


@given("A Fediverse server")
def get_fediverse_actor(context):
    webfinger = get_webfinger()
    actor_url = None
    for link in webfinger["links"]:
        if "type" in link:
            if all(
                [link["rel"] == "self", link["type"] == "application/activity+json"]
            ):
                actor_url = link["href"]

    assert actor_url is not None
    logger.info(f"Actor URL {actor_url}")
    context.actor_url = actor_url


@when("Receiving or querying an object")
def check_actor_exists(context):
    actor = get_ap_obj(context.actor_url)
    assert actor is not None
    context.actor = actor


@then("the response must contain an 'id' and a 'type' parameter")
def verify_obj_attrs(context):
    name = "verify_obj_attrs"
    actor = context.actor
    try:
        assert "id" in actor, "'id' attribute present in object'"
        assert "type" in actor, "'type' attribute present in object'"
    except Exception as e:
        logger.error(e)
        context.failure[name] = e
        raise e

    context.success.append(name)


@then("The 'id' must resolve to the same object")
def verify_id_resolves(context):
    name = "verify_id_resolves"
    actor = context.actor
    try:
        assert get_ap_obj(actor["id"]) == actor, "'id' URL resolves to same object"
    except Exception as e:
        logger.error(e)
        context.failure[name] = e
        raise e
    context.success.append(name)


@then("context must be ActivityPub")
def ctx_is_ap(context):
    name = "ctx_is_ap"

    obj_ctx = context.actor["@context"]
    activitypub_present = False
    for ctx in obj_ctx:
        if ctx == "https://www.w3.org/ns/activitystreams":
            activitypub_present = True
            break
    try:
        assert activitypub_present is True, "ActivityPub context is included"
    except Exception as e:
        logger.error(e)
        context.failure[name] = e
        raise e
    context.success.append(name)