python-mapswipe-workers/mapswipe_workers/tests/integration/tear_down.py at e2efb18a07c60e3aa28936863e30cc768d67fee8 · mapswipe/python-mapswipe-workers · 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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
"""Helper functions for test tear down"""
import os
import re
import tempfile
import time
from typing import List

from mapswipe_workers import auth


def delete_test_data(project_id: str, tutorial_id: str = None) -> None:
    """
    Delete test project indluding groups, tasks and results
    from Firebase and Postgres
    """

    if not re.match(r"[-a-zA-Z0-9]+", project_id):
        raise ValueError(
            "Given argument resulted in invalid Firebase Realtime Database reference. "
        )

    fb_db = auth.firebaseDB()
    ref = fb_db.reference(f"v2/results/{project_id}")
    ref.delete()
    ref = fb_db.reference(f"v2/tasks/{project_id}")
    ref.delete()
    ref = fb_db.reference(f"v2/groupsUsers/{project_id}")
    ref.delete()
    time.sleep(1)  # Wait for Firebase Functions to complete
    ref = fb_db.reference(f"v2/groups/{project_id}")
    ref.delete()
    ref = fb_db.reference(f"v2/projects/{project_id}")
    ref.delete()
    ref = fb_db.reference(f"v2/projectDrafts/{project_id}")
    ref.delete()
    ref = fb_db.reference(f"v2/tutorialDrafts/{project_id}")
    ref.delete()
    ref = fb_db.reference(f"v2/users/{project_id}")
    ref.delete()

    if tutorial_id is not None:
        ref = fb_db.reference(f"v2/projects/{tutorial_id}")
        ref.delete()
        ref = fb_db.reference(f"v2/tutorialDrafts/{tutorial_id}")
        ref.delete()

    # Clear out the user-group used in test.
    # XXX: Use a firebase simulator for running test.
    # For CI/CD, use a real firebase with scope using commit hash,
    # and clear all data at the end.
    for user_group_id in [
        "dummy-user-group-1",
        "dummy-user-group-2",
        "dummy-user-group-3",
        "dummy-user-group-4",
    ]:
        ref = fb_db.reference(f"v2/userGroups/{user_group_id}")
        ref.delete()

    pg_db = auth.postgresDB()
    # Delete user results data
    sql_query = (
        "DELETE FROM mapping_sessions_results "
        "WHERE mapping_session_id IN ("
        "SELECT mapping_session_id "
        "FROM mapping_sessions WHERE project_id = %s)"
    )
    pg_db.query(sql_query, [project_id])
    sql_query = (
        "DELETE FROM mapping_sessions_results_geometry "
        "WHERE mapping_session_id IN ("
        "SELECT mapping_session_id "
        "FROM mapping_sessions WHERE project_id = %s)"
    )
    pg_db.query(sql_query, [project_id])
    # Delete user-groups results data
    sql_query = (
        "DELETE FROM mapping_sessions_user_groups "
        "WHERE mapping_session_id IN ("
        "SELECT mapping_session_id "
        "FROM mapping_sessions WHERE project_id = %s)"
    )
    pg_db.query(sql_query, [project_id])
    # Delete mapping sessions
    sql_query = "DELETE FROM mapping_sessions WHERE project_id = %s"
    pg_db.query(sql_query, [project_id])
    sql_query = "DELETE FROM results_temp WHERE project_id = %s"
    pg_db.query(sql_query, [project_id])
    sql_query = "DELETE FROM results_geometry_temp WHERE project_id = %s"
    pg_db.query(sql_query, [project_id])
    sql_query = "DELETE FROM tasks WHERE project_id = %s"
    pg_db.query(sql_query, [project_id])
    sql_query = "DELETE FROM groups WHERE project_id = %s"
    pg_db.query(sql_query, [project_id])
    sql_query = "DELETE FROM projects WHERE project_id = %s"
    pg_db.query(sql_query, [project_id])

    sql_query = "DELETE FROM users WHERE user_id = 'test_build_area'"
    pg_db.query(sql_query)
    sql_query = "DELETE FROM users_temp WHERE user_id = 'test_build_area'"
    pg_db.query(sql_query)

    sql_query = "DELETE FROM users WHERE user_id = 'test_build_area_heidelberg'"
    pg_db.query(sql_query)
    sql_query = "DELETE FROM users_temp WHERE user_id = 'test_build_area_heidelberg'"
    pg_db.query(sql_query)

    filename = os.path.join(
        tempfile._get_default_tempdir(), f"results_{project_id}.csv.gz"
    )
    try:
        os.remove(filename)
    except FileNotFoundError:
        pass


def delete_test_user_group(user_group_ids: List) -> None:
    # Make sure delete_test_data is runned first.
    fb_db = auth.firebaseDB()
    ref = fb_db.reference("v2/groupsUsers")
    ref.delete()

    pg_db = auth.postgresDB()
    pg_db.query(
        "DELETE FROM user_groups_user_memberships WHERE user_group_id = ANY(%s);",
        [user_group_ids],
    )
    pg_db.query(
        "DELETE FROM user_groups WHERE user_group_id = ANY(%s);",
        [user_group_ids],
    )


def delete_test_user(user_ids: List) -> None:
    for user_id in user_ids:
        if not re.match(r"[-a-zA-Z0-9]+", user_id):
            raise ValueError(
                "Given argument resulted in invalid "
                "Firebase Realtime Database reference. "
            )

        fb_db = auth.firebaseDB()
        ref = fb_db.reference(f"v2/users/{user_id}")
        ref.delete()

    pg_db = auth.postgresDB()
    sql_query = "DELETE FROM users WHERE user_id = ANY( %(user_ids)s );"
    pg_db.query(sql_query, {"user_ids": user_ids})
    sql_query = "DELETE FROM users_temp WHERE user_id = ANY( %(user_ids)s );"
    pg_db.query(sql_query, {"user_ids": user_ids})


if __name__ == "__main__":
    delete_test_data("test_build_area")
    delete_test_data("test_build_area_heidelberg")
    delete_test_data("test_tile_classification")