test_processors.py

# -*- coding: utf-8 -*-
#
# This file is part of Invenio.
# Copyright (C) 2018 CERN.
# Copyright (C) 2025 Graz University of Technology.
#
# Invenio is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.

"""Event processor tests."""

import logging
from datetime import datetime
from unittest.mock import patch

import pytest
from conftest import _create_file_download_event
from helpers import get_queue_size, mock_date
from invenio_queues.proxies import current_queues
from invenio_search import current_search
from invenio_search.engine import dsl

from invenio_stats.contrib.event_builders import (
    build_file_unique_id,
    file_download_event_builder,
)
from invenio_stats.processors import (
    EventsIndexer,
    anonymize_user,
    flag_machines,
    flag_robots,
    hash_id,
)
from invenio_stats.proxies import current_stats
from invenio_stats.tasks import process_events


@pytest.mark.parametrize(
    [
        "ip_addess",
        "user_id",
        "session_id",
        "user_agent",
        "timestamp",
        "exp_country",
        "exp_visitor_id",
        "exp_unique_session_id",
    ],
    [
        # Minimal
        (
            "131.169.180.47",
            None,
            None,
            None,
            datetime(2018, 1, 1, 12),
            "DE",
            "1850abef504ce64cb6b38fa60fe8f90aede1d2d2e9013735554af946",
            "1850abef504ce64cb6b38fa60fe8f90aede1d2d2e9013735554af946",
        ),
        # User id
        (
            "188.184.37.205",
            "100",
            None,
            None,
            datetime(2018, 1, 1, 12),
            "CH",
            "eaf6a44a598ea63c659e6e46722e2d11d0d7487694ec504ade273b9d",
            "c6b85f117cd0636a07f1cf250a30d86714ec45e55a1110441d1a9e2b",
        ),
        # User id + session id + user agent, different IP address
        (
            "23.22.39.120",
            "100",
            "foo",
            "bar",
            datetime(2018, 1, 1, 12),
            "US",
            "eaf6a44a598ea63c659e6e46722e2d11d0d7487694ec504ade273b9d",
            "c6b85f117cd0636a07f1cf250a30d86714ec45e55a1110441d1a9e2b",
        ),
        # User id, different hour
        (
            "23.22.39.120",
            "100",
            None,
            None,
            datetime(2018, 1, 1, 15),
            "US",
            "eaf6a44a598ea63c659e6e46722e2d11d0d7487694ec504ade273b9d",
            "77536991d991e6e8251999fc6a8d78ec1be42847da3c8774221a03a0",
        ),
        # User id, same hour different minute
        (
            "23.22.39.120",
            "100",
            None,
            None,
            datetime(2018, 1, 1, 15, 30),
            "US",
            "eaf6a44a598ea63c659e6e46722e2d11d0d7487694ec504ade273b9d",
            "77536991d991e6e8251999fc6a8d78ec1be42847da3c8774221a03a0",
        ),
        # Session id
        (
            "131.169.180.47",
            None,
            "foo",
            None,
            datetime(2018, 1, 1, 12),
            "DE",
            "a78cc092c88fb4d060a873217f2cd466c2776f672a99ee06317c2858",
            "ca28702a6ece34d18c6f6498ef79d77492a6bd653ac886beb5018880",
        ),
        # Session id + user agent
        (
            "131.169.180.47",
            None,
            "foo",
            "bar",
            datetime(2018, 1, 1, 12),
            "DE",
            "a78cc092c88fb4d060a873217f2cd466c2776f672a99ee06317c2858",
            "ca28702a6ece34d18c6f6498ef79d77492a6bd653ac886beb5018880",
        ),
        # Session id + user agent + different hour
        (
            "131.169.180.47",
            None,
            "foo",
            "bar",
            datetime(2018, 1, 1, 15),
            "DE",
            "a78cc092c88fb4d060a873217f2cd466c2776f672a99ee06317c2858",
            "ceb752c8b51c8c4a9c18a9d4404e9fb570fbf83195631ab3efb46b31",
        ),
        # User agent
        (
            "188.184.37.205",
            None,
            None,
            "bar",
            datetime(2018, 1, 1, 12),
            "CH",
            "e9c48686d21c21a9ee5b9eba58b1d86c9460272809b6de71649f6ce7",
            "e9c48686d21c21a9ee5b9eba58b1d86c9460272809b6de71649f6ce7",
        ),
        # Differnet ip address
        (
            "131.169.180.47",
            None,
            None,
            "bar",
            datetime(2018, 1, 1, 12),
            "DE",
            "602e9bc738b422d5a19283e20fc31ec540a12d42b04ad7073d943fb2",
            "602e9bc738b422d5a19283e20fc31ec540a12d42b04ad7073d943fb2",
        ),
        # Different hour
        (
            "131.169.180.47",
            None,
            None,
            "bar",
            datetime(2018, 1, 1, 15),
            "DE",
            "4b30c060f422f304b073759553d4161a14784e0ddcf57284f55d7cae",
            "4b30c060f422f304b073759553d4161a14784e0ddcf57284f55d7cae",
        ),
        # No result ip address
        (
            "0.0.0.0",
            None,
            None,
            None,
            datetime(2018, 1, 1, 12),
            None,
            "1850abef504ce64cb6b38fa60fe8f90aede1d2d2e9013735554af946",
            "1850abef504ce64cb6b38fa60fe8f90aede1d2d2e9013735554af946",
        ),
    ],
)
def test_anonymize_user(
    mock_anonymization_salt,
    ip_addess,
    user_id,
    session_id,
    user_agent,
    timestamp,
    exp_country,
    exp_visitor_id,
    exp_unique_session_id,
):
    """Test anonymize_user preprocessor."""
    event = anonymize_user(
        {
            "ip_address": ip_addess,
            "user_id": user_id,
            "session_id": session_id,
            "user_agent": user_agent,
            "timestamp": timestamp.isoformat(),
        }
    )
    assert "user_id" not in event
    assert "user_agent" not in event
    assert "ip_address" not in event
    assert "session_id" not in event
    assert event["country"] == exp_country
    assert event["visitor_id"] == exp_visitor_id
    assert event["unique_session_id"] == exp_unique_session_id


def test_anonymiation_salt(app):
    """Test anonymization salt for different days."""
    event = anonymize_user(
        {
            "ip_address": "131.169.180.47",
            "user_id": "100",
            "timestamp": datetime(2018, 1, 1, 12).isoformat(),
        }
    )
    event_same_day = anonymize_user(
        {
            "ip_address": "131.169.180.47",
            "user_id": "100",
            "timestamp": datetime(2018, 1, 1, 21).isoformat(),
        }
    )
    event_other_day = anonymize_user(
        {
            "ip_address": "131.169.180.47",
            "user_id": "100",
            "timestamp": datetime(2018, 1, 2, 12).isoformat(),
        }
    )

    # Same user, same day -> identical visitor id
    assert event["visitor_id"] == event_same_day["visitor_id"]
    # Same user, same day, different hour -> different unique session id
    assert event["unique_session_id"] != event_same_day["unique_session_id"]
    # Same user, different day -> different visitor id
    assert event["visitor_id"] != event_other_day["visitor_id"]
    # Same user, different day and hour -> different unique session id
    assert event["unique_session_id"] != event_other_day["unique_session_id"]


def test_flag_robots(app, mock_user_ctx, request_headers, objects):
    """Test flag_robots preprocessor."""

    def build_event(headers):
        with app.test_request_context(headers=headers):
            event = file_download_event_builder({}, app, objects[0])
        return flag_robots(event)

    assert build_event(request_headers["user"])["is_robot"] is False
    assert build_event(request_headers["machine"])["is_robot"] is True
    # see https://github.com/inveniosoftware/counter-robots/issues/15
    # assert build_event(request_headers["robot"])["is_robot"] is False


def test_flag_machines(app, mock_user_ctx, request_headers, objects):
    """Test machines preprocessor."""

    def build_event(headers):
        with app.test_request_context(headers=headers):
            event = file_download_event_builder({}, app, objects[0])
        return flag_machines(event)

    assert build_event(request_headers["user"])["is_machine"] is False
    assert build_event(request_headers["robot"])["is_machine"] is False
    assert build_event(request_headers["machine"])["is_machine"] is True


def test_referrer(app, mock_user_ctx, request_headers, objects):
    """Test referrer header."""
    request_headers["user"]["REFERER"] = "example.com"
    with app.test_request_context(headers=request_headers["user"]):
        event = file_download_event_builder({}, app, objects[0])
    assert event["referrer"] == "example.com"


def test_events_indexer_preprocessors(app, mock_event_queue):
    """Check that EventsIndexer calls properly the preprocessors."""

    def test_preprocessor1(event):
        event["test1"] = 42
        event["visitor_id"] = "testuser1"
        return event

    def test_preprocessor2(event):
        event["test2"] = 21
        return event

    indexer = EventsIndexer(
        mock_event_queue,
        preprocessors=[build_file_unique_id, test_preprocessor1, test_preprocessor2],
    )

    # Generate the events
    received_docs = []

    def bulk(client, generator, *args, **kwargs):
        received_docs.extend(generator)

    with patch("invenio_stats.processors.datetime", mock_date(2017, 6, 2, 12)):
        with patch("invenio_search.engine.search.helpers.bulk", side_effect=bulk):
            indexer.run()

    # Process the events as we expect them to be
    expected_docs = []
    for event in mock_event_queue.queued_events:
        event = build_file_unique_id(event)
        event = test_preprocessor1(event)
        event = test_preprocessor2(event)
        event["updated_timestamp"] = "2017-06-02T12:00:00+00:00"
        _id = hash_id("2017-01-01T00:00:00", event)
        expected_docs.append(
            {
                "_id": _id,
                "_op_type": "index",
                "_index": "events-stats-file-download-2017-01",
                "_source": event,
            }
        )
    assert received_docs == expected_docs


def test_events_indexer_id_windowing(app, mock_event_queue):
    """Check that EventsIndexer applies time windows to ids."""

    indexer = EventsIndexer(mock_event_queue, preprocessors=[], double_click_window=180)

    # Generated docs will be registered in this list
    received_docs = []

    def bulk(client, generator, *args, **kwargs):
        received_docs.extend(generator)

    mock_event_queue.consume.return_value = [
        _create_file_download_event(date)
        for date in [
            # Those two events will be in the same window
            (2017, 6, 1, 0, 11, 3),
            (2017, 6, 1, 0, 9, 1),
            # Those two events will be in the same window
            (2017, 6, 2, 0, 12, 10),
            (2017, 6, 2, 0, 13, 3),
            (2017, 6, 2, 0, 30, 3),
        ]
    ]

    with patch("invenio_search.engine.search.helpers.bulk", side_effect=bulk):
        indexer.run()

    assert len(received_docs) == 5
    ids = set(doc["_id"] for doc in received_docs)
    assert len(ids) == 3


def test_double_clicks(app, mock_event_queue, search_clear):
    """Test that events occurring within a time window are counted as 1."""
    event_type = "file-download"
    events = [
        _create_file_download_event(date)
        for date in [
            (2000, 6, 1, 10, 0, 10),
            (2000, 6, 1, 10, 0, 11),
            (2000, 6, 1, 10, 0, 19),
            (2000, 6, 1, 10, 0, 22),
        ]
    ]
    current_queues.declare()
    current_stats.publish(event_type, events)
    process_events(["file-download"])
    current_search.flush_and_refresh(index="*")
    res = search_clear.search(
        index="events-stats-file-download-2000-06",
    )
    assert res["hits"]["total"]["value"] == 2


def test_failing_processors(app, search_clear, event_queues, caplog):
    """Test events that raise an exception when processed."""
    search_obj = dsl.Search(using=search_clear)

    current_queues.declare()
    current_stats.publish(
        "file-download",
        [
            _create_file_download_event(date)
            for date in [(2018, 1, 1), (2018, 1, 2), (2018, 1, 3), (2018, 1, 4)]
        ],
    )

    def _raises_on_second_call(doc):
        if _raises_on_second_call.calls == 1:
            _raises_on_second_call.calls += 1
            raise Exception("mocked-exception")
        _raises_on_second_call.calls += 1
        return doc

    _raises_on_second_call.calls = 0

    queue = current_queues.queues["stats-file-download"]
    indexer = EventsIndexer(queue, preprocessors=[_raises_on_second_call])

    current_search.flush_and_refresh(index="*")
    assert get_queue_size("stats-file-download") == 4
    assert not search_clear.indices.exists("events-stats-file-download-2018-01")
    assert not search_clear.indices.exists("events-stats-file-download-2018-01")
    assert not search_clear.indices.exists("events-stats-file-download-2018-01")
    assert not search_clear.indices.exists("events-stats-file-download-2018-01")
    assert not search_clear.indices.exists_alias(name="events-stats-file-download")

    with caplog.at_level(logging.ERROR):
        indexer.run()  # 2nd event raises exception and is dropped

    # Check that the error was logged
    error_logs = [r for r in caplog.records if r.levelno == logging.ERROR]
    assert len(error_logs) == 1
    assert error_logs[0].msg == "Error while processing event"
    assert error_logs[0].exc_info[1].args[0] == "mocked-exception"

    current_search.flush_and_refresh(index="*")
    assert get_queue_size("stats-file-download") == 0
    assert search_obj.index("events-stats-file-download").count() == 3
    assert search_obj.index("events-stats-file-download-2018-01").count() == 3


BROWSER_UA = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36"
)


def _classifier(asn_by_ip=None):
    """Build a real counter-robots classifier (COUNTER baseline + extended preset)."""
    from counter_robots import ClassifierBuilder, counter_preset, extended_preset

    builder = ClassifierBuilder().use(counter_preset).use(extended_preset)
    if asn_by_ip is not None:
        builder.asn_resolver(lambda ip: asn_by_ip.get(ip))
    return builder.build()


def _with_visitor_classifier(classifier):
    """App context exposing ``current_stats.visitor_classifier``."""
    from types import SimpleNamespace

    from flask import Flask

    app = Flask(__name__)
    app.extensions["invenio-stats"] = SimpleNamespace(visitor_classifier=classifier)
    return app.app_context()


def test_flag_robots_uses_classifier():
    """flag_robots goes through the visitor classifier (extended preset here)."""
    import invenio_stats.processors as procs

    with _with_visitor_classifier(_classifier()):
        # CamelCase bot the case-sensitive COUNTER baseline misses
        assert procs.flag_robots({"user_agent": "YisouSpider"})["is_robot"] is True
        assert procs.flag_robots({"user_agent": BROWSER_UA})["is_robot"] is False


def test_flag_machines_uses_classifier():
    import invenio_stats.processors as procs

    with _with_visitor_classifier(_classifier()):
        ua = {"user_agent": "Python/3.10 aiohttp/3.10.5"}
        assert procs.flag_machines(ua)["is_machine"] is True
        assert procs.flag_machines({"user_agent": BROWSER_UA})["is_machine"] is False


def test_exclude_datacenter_browser():
    """Browser events from datacenter IPs are dropped; others pass unchanged."""
    import invenio_stats.processors as procs

    # AS16509 AWS (datacenter), AS714 Apple (allow-listed), AS3320 Deutsche Telekom.
    asn_by_ip = {"1.1.1.1": 16509, "2.2.2.2": 714, "3.3.3.3": 3320}
    with _with_visitor_classifier(_classifier(asn_by_ip)):

        def excl(ip, ua):
            return procs.exclude_datacenter_browser(
                {"ip_address": ip, "user_agent": ua}
            )

        assert excl("1.1.1.1", BROWSER_UA) is None  # browser from AWS -> dropped
        # everything else passes through unchanged, with no field written
        kept = excl("3.3.3.3", BROWSER_UA)  # eyeball ISP
        assert kept is not None and "is_datacenter" not in kept
        assert excl("2.2.2.2", BROWSER_UA) is not None  # Apple, allow-listed
        assert excl("1.1.1.1", "python-requests/2.31") is not None  # not a browser
        assert excl(None, BROWSER_UA) is not None  # no IP


def test_exclude_datacenter_browser_noop_without_resolver():
    """With no ASN resolver, the datacenter check is a no-op and writes no field."""
    import invenio_stats.processors as procs

    with _with_visitor_classifier(_classifier()):
        doc = {"ip_address": "1.1.1.1", "user_agent": BROWSER_UA}
        result = procs.exclude_datacenter_browser(doc)
        assert result is doc
        assert "is_datacenter" not in result