mobility-feed-api/functions-python/update_feed_status/tests/test_update_feed_status_main.py at 1070aaa702780f8be43617c73881ee96f109cd69 · MobilityData/mobility-feed-api · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
from unittest.mock import patch, MagicMock

from conftest import clean_testing_db, populate_database
from shared.database.database import with_db_session
from test_shared.test_utils.database_utils import default_db_url
from main import (
    update_feed_status,
    update_feed_statuses_query,
)
from shared.database_gen.sqlacodegen_models import Feed
from datetime import date, timedelta
from typing import Iterator, NamedTuple

import os

from sqlalchemy import text
from sqlalchemy.orm import Session


class PartialFeed(NamedTuple):
    """
    Subset of the Feed entity with only the fields queried in `fetch_feeds`.
    """

    id: str
    status: str


def fetch_feeds(session: Session) -> Iterator[PartialFeed]:
    # When adding or removing fields here, `PartialFeed` should be updated to
    # match, for type safety.
    query = session.query(Feed.id, Feed.status).filter(
        Feed.status != text("'deprecated'::status"),
        Feed.status != text("'development'::status"),
    )
    for feed in query:
        yield PartialFeed(id=feed.id, status=feed.status)


@with_db_session(db_url=default_db_url)
def test_update_feed_status(db_session: Session) -> None:
    feeds_before: dict[str, PartialFeed] = {f.id: f for f in fetch_feeds(db_session)}
    result = dict(update_feed_statuses_query(db_session, []))
    assert result == {
        "inactive": 3,
        "active": 2,
        "future": 1,
    }

    feeds_after: dict[str, PartialFeed] = {f.id: f for f in fetch_feeds(db_session)}
    expected_status_changes = {
        "2": "active",
        "7": "inactive",
        "8": "inactive",
        "10": "future",
        "22": "inactive",
        "25": "active",
    }
    for feed_id, feed_before in feeds_before.items():
        feed_after = feeds_after[feed_id]
        assert feed_after.status == expected_status_changes.get(
            feed_id, feed_before.status
        )


@with_db_session(db_url=default_db_url)
def test_update_feed_status_with_ids(db_session: Session) -> None:
    clean_testing_db()
    populate_database()
    feeds_before: dict[str, PartialFeed] = {f.id: f for f in fetch_feeds(db_session)}
    result = dict(update_feed_statuses_query(db_session, ["mdb-8"]))
    assert result == {
        "inactive": 1,
        "active": 0,
        "future": 0,
    }

    feeds_after: dict[str, PartialFeed] = {f.id: f for f in fetch_feeds(db_session)}
    expected_status_changes = {
        "8": "inactive",
    }
    for feed_id, feed_before in feeds_before.items():
        feed_after = feeds_after[feed_id]
        assert feed_after.status == expected_status_changes.get(
            feed_id, feed_before.status
        )


def test_update_feed_status_failed_query():
    mock_session = MagicMock()

    today = date(2025, 3, 1)

    mock_subquery = MagicMock()
    mock_subquery.c.feed_id = 1
    mock_subquery.c.service_date_range_start = today - timedelta(days=10)
    mock_subquery.c.service_date_range_end = today + timedelta(days=10)

    mock_query = mock_session.query.return_value
    mock_query.filter.return_value.subquery.return_value = mock_subquery

    mock_update_query = mock_session.query.return_value.filter.return_value
    mock_update_query.update.side_effect = Exception("Mocked exception")

    try:
        update_feed_statuses_query(mock_session, [])
    except Exception as e:
        assert str(e) == "Error updating feed statuses: Mocked exception"


@patch("main.update_feed_statuses_query")
def test_updated_feed_status(mock_update_query):
    return_value = {"active": 5}
    mock_update_query.return_value = return_value

    with patch.dict(os.environ, {"FEEDS_DATABASE_URL": default_db_url}):
        response_body, status_code = update_feed_status(None)

    mock_update_query.assert_called_once()
    assert response_body == return_value
    assert status_code == 200


@patch("main.update_feed_statuses_query")
def test_updated_feed_status_error_raised(mock_update_query):
    mock_update_query.side_effect = Exception("Mocked exception")

    with patch.dict(os.environ, {"FEEDS_DATABASE_URL": default_db_url}):
        response_body, status_code = update_feed_status(None)

    mock_update_query.assert_called_once()
    assert response_body == "Error updating the feed statuses: Mocked exception"
    assert status_code == 500