aboutsummaryrefslogtreecommitdiff
path: root/tests/testutils.py
blob: 9db3673b8b52b579865cd6e6b1f409ba5268ae93 (plain)
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
"""Various utility functions for testing."""
import datetime
import gzip
import io
from pathlib import Path
from typing import NamedTuple

from playwright.sync_api import Page
from sqlalchemy import Engine
from sqlalchemy.orm import Session

from fietsboek import convert, models, util
from fietsboek.data import DataManager


class PopulationIds(NamedTuple):
    """Collection of database IDs that :func:`populate` returns."""
    jon: int
    davos: int


def load_test_asset(filename: str) -> bytes:
    """Load a test asset.

    Unlike :func:`load_gpx_asset`, this function does not do gzip decompression.

    :param filkename: Name of the asset to load.
    :return: The content of the file as bytes.
    """
    asset_dir = Path(__file__).parent / "assets"
    return (asset_dir / filename).read_bytes()


def load_gpx_asset(filename: str) -> bytes:
    """Load a GPX test asset.

    This looks in the tests/assets/ folder, reads and unzips the file and
    returns its contents.

    :param filename: Name of the asset to load.
    :return: The content of the asset as bytes.
    """
    asset_dir = Path(__file__).parent / 'assets'
    test_file = asset_dir / filename
    with gzip.open(test_file, 'rb') as fobj:
        return fobj.read()


def extract_and_upload(page: Page, filename: str, tmp_path: Path):
    """Extracts the given test asset, fills in the upload form and presses
    upload.

    :param page: The playwright page on which to execute the actions.
    :param filename: The filename.
    :param tmp_path: The temporary path (as given by pytest).
    """
    gpx_data = load_gpx_asset(filename)
    gpx_path = tmp_path / "Upload.gpx"
    with open(gpx_path, "wb") as gpx_fobj:
        gpx_fobj.write(gpx_data)

    page.get_by_label("GPX file").set_input_files(gpx_path)

    page.locator(".bi-upload").click()


def populate(dbengine: Engine, data_manager: DataManager) -> PopulationIds:
    """Populates the database and data directory with some test data."""
    user_ids = []
    with Session(dbengine) as session:
        user = models.User(name="Jon", email="jon.snow@nw.org", is_verified=True, is_admin=True)
        user.set_password("ygritte")
        user.roll_session_secret()
        session.add(user)
        session.flush()
        user_ids.append(user.id)

        user = models.User(name="Davos", email="davos@seaworth.com", is_verified=True)
        user.set_password("123456")
        user.roll_session_secret()
        session.add(user)
        session.flush()
        user_ids.append(user.id)

        session.commit()

    for user_id in user_ids:
        data_manager.initialize_user(user_id)

    # First track for Jon:

    gpx_data = load_gpx_asset("Teasi_1.gpx.gz")
    track = convert.smart_convert(gpx_data)
    path = track.path()
    track.points = []
    track.owner_id = user_ids[0]
    track.title = "Trip around Winterfell"
    track.visibility = models.track.Visibility.PUBLIC
    track.type = models.track.TrackType.ORGANIC
    track.description = "I took my sister for a quick trip around Winterfell"
    track.badges = []
    track.link_secret = util.random_link_secret()
    track.tagged_people = []
    track.date = datetime.datetime(1984, 1, 2, 10, 11, tzinfo=datetime.UTC)
    track.transformers = []
    track.sync_tags({"westeros"})

    with Session(dbengine) as session:
        session.add(track)
        session.flush()

        assert track.id is not None
        manager = data_manager.initialize(track.id)
        manager.compress_backup(gpx_data)

        track.fast_set_path(path)
        track.ensure_cache(path)
        session.add(track.cache)

        image_name = manager.add_image(io.BytesIO(load_test_asset("picture01.jpg")), "PIC001.jpg")
        image_meta = models.ImageMetadata(track=track, image_name=image_name)
        image_meta.description = "Beautiful sight out of the Hunter's Gate"
        session.add(image_meta)

        image_name = manager.add_image(io.BytesIO(load_test_asset("picture02.jpg")), "PIC002.jpg")
        image_meta = models.ImageMetadata(track=track, image_name=image_name)
        image_meta.description = "Our steel wire horses ready to gallop"
        session.add(image_meta)

        session.commit()

    # Second track for Jon:

    gpx_data = load_gpx_asset("MyTourbook_1.gpx.gz")
    track = convert.smart_convert(gpx_data)
    path = track.path()
    track.points = []
    track.owner_id = user_ids[0]
    track.title = "Road to Riverrun"
    track.visibility = models.track.Visibility.PUBLIC
    track.type = models.track.TrackType.ORGANIC
    track.description = "Got a wedding to attend!"
    track.badges = []
    track.link_secret = util.random_link_secret()
    track.tagged_people = []
    track.date = datetime.datetime(1985, 8, 5, 16, 41, tzinfo=datetime.UTC)
    track.transformers = []
    track.sync_tags({"westeros"})

    with Session(dbengine) as session:
        session.add(track)
        session.flush()

        assert track.id is not None
        manager = data_manager.initialize(track.id)
        manager.compress_backup(gpx_data)

        track.fast_set_path(path)
        track.ensure_cache(path)
        session.add(track.cache)

        session.commit()

    return PopulationIds(jon=user_ids[0], davos=user_ids[1])