-
Notifications
You must be signed in to change notification settings - Fork 2
/
test_utils.py
96 lines (76 loc) · 2.58 KB
/
test_utils.py
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
import os
from pathlib import Path, PosixPath
import pytest
import requests
from mwsql import head, load
from mwsql.utils import _open_file
from .helpers import Capturing
# from urllib.error import HTTPError
CURRENT_DIR = Path(__file__).parent
DATA_DIR = CURRENT_DIR.parent / "data"
FILEPATH_GZ = DATA_DIR / "testfile.sql.gz"
FILEPATH_UNZIPPED = DATA_DIR / "testfile.sql"
@pytest.mark.parametrize(
"database,filename,date,extension,expected",
[
(
"simplewiki",
"change_tag_def",
"latest",
"sql",
"simplewiki-latest-change_tag_def.sql.gz",
),
(
"simplewiki",
"change_tag",
"latest",
"sql",
"simplewiki-latest-change_tag.sql.gz",
),
("bewiki", "site_stats", "latest", "sql", "bewiki-latest-site_stats.sql.gz"),
("nlwiktionary", "sites", "latest", "sql", "nlwiktionary-latest-sites.sql.gz"),
],
)
def test_load(database, filename, date, extension, expected):
f = load(database, filename, date, extension)
assert f == PosixPath(expected)
os.remove(f)
def test_load_HTTPError():
with pytest.raises(requests.exceptions.HTTPError):
load("simplewiki", "non-existing-filename", "latest")
def test__open_file_gz():
with _open_file(FILEPATH_GZ) as infile:
for line in infile:
assert (
line
== "-- MySQL dump 10.18 Distrib 10.3.27-MariaDB, for debian-linux-gnu (x86_64)\n"
)
break
def test__open_file_unzipped():
with _open_file(FILEPATH_UNZIPPED) as infile:
for line in infile:
assert (
line
== "-- MySQL dump 10.18 Distrib 10.3.27-MariaDB, for debian-linux-gnu (x86_64)\n"
)
break
expected_out = [
"-- MySQL dump 10.18 Distrib 10.3.27-MariaDB, for debian-linux-gnu (x86_64)",
"--",
"-- Host: 10.64.32.82 Database: simplewiki",
"-- ------------------------------------------------------",
"-- Server version\t10.4.19-MariaDB-log",
"",
"/*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;",
"/*!40101 SET @OLD_CHARACTER_SET_RESULTS=@@CHARACTER_SET_RESULTS */;",
"/*!40101 SET @OLD_COLLATION_CONNECTION=@@COLLATION_CONNECTION */;",
"/*!40101 SET NAMES utf8mb4 */;",
]
def test_head_gz():
with Capturing() as output:
head(FILEPATH_GZ, 10)
assert output == expected_out
def test_head_unzipped():
with Capturing() as output:
head(FILEPATH_UNZIPPED, 10)
assert output == expected_out