Add parameter for selecting specific namespaces.
[mediawiki_dump_tools.git] / test / Wikiq_Unit_Test.py
1 import unittest
2 import os
3 import subprocess
4 from shutil import copyfile
5 import pandas as pd
6 from pandas.util.testing import assert_frame_equal
7 from io import StringIO
8
9 # with / without pwr DONE
10 # with / without url encode DONE
11 # with / without collapse user DONE
12 # with output to sdtout DONE
13 # note that the persistence radius is 7 by default
14 # reading various file formats including
15 #        7z, gz, bz2, xml  DONE
16 # wikia and wikipedia data DONE
17 # malformed xmls DONE
18
19 class Test_Wikipedia(unittest.TestCase):
20     def setUp(self):
21         if not os.path.exists("test_output"):
22             os.mkdir("test_output")
23
24         self.wiki = 'ikwiki-20180301-pages-meta-history'
25         self.wikiq_out_name =  self.wiki + ".tsv"
26         self.test_output_dir = os.path.join(".", "test_output")
27         self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name)
28
29         self.infile = "{0}.xml.bz2".format(self.wiki)    
30         self.base_call = "../wikiq {0} -o {1}"
31         self.input_dir = "dumps"
32         self.input_file = os.path.join(".", self.input_dir,self.infile)
33         self.baseline_output_dir = "baseline_output"
34
35     def test_WP_url_encode(self):
36         test_filename =  "url-encode_" + self.wikiq_out_name
37         test_file = os.path.join(self.test_output_dir, test_filename)
38         if os.path.exists(test_file):
39             os.remove(test_file)
40         
41         call = self.base_call.format(self.input_file, self.test_output_dir)
42         call = call + " --url-encode"
43         proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True)
44         proc.wait()
45
46         copyfile(self.call_output, test_file)
47         baseline_file = os.path.join(".", self.baseline_output_dir, test_filename)
48
49         # as a test let's make sure that we get equal data frames
50         test = pd.read_table(test_file)
51         baseline = pd.read_table(baseline_file)
52         assert_frame_equal(test,baseline)
53
54     def test_WP_namespaces(self):
55         print(os.path.abspath('.'))
56         test_filename =  "namespaces_" + self.wikiq_out_name
57         test_file = os.path.join(self.test_output_dir, test_filename)
58         if os.path.exists(test_file):
59             os.remove(test_file)
60         
61         call = self.base_call.format(self.input_file, self.test_output_dir)
62         call = call + " -n 0 -n 1"
63         print(call)
64         proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True)
65         proc.wait()
66         copyfile(self.call_output, test_file)
67         baseline_file = os.path.join(os.path.abspath("."), self.baseline_output_dir, test_filename)
68
69         # as a test let's make sure that we get equal data frames
70         test = pd.read_table(test_file)
71         num_wrong_ns = sum(~ test.namespace.isin({0,1}))
72         self.assertEquals(num_wrong_ns, 0)
73         baseline = pd.read_table(baseline_file)
74         assert_frame_equal(test,baseline)
75
76
77 class Test_Basic(unittest.TestCase):
78
79     def setUp(self):
80         if not os.path.exists("test_output"):
81             os.mkdir("test_output")
82
83         self.wiki = 'sailormoon'
84         self.wikiq_out_name =  self.wiki + ".tsv"
85         self.test_output_dir = os.path.join(".", "test_output")
86         self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name)
87
88         self.infile = "{0}.xml.7z".format(self.wiki)
89         self.base_call = "../wikiq {0} -o {1}"
90         self.input_dir = "dumps"
91         self.input_file = os.path.join(".", self.input_dir,self.infile)
92         self.baseline_output_dir = "baseline_output"
93
94     def test_noargs(self):
95
96         test_filename =  "noargs_" + self.wikiq_out_name
97         test_file = os.path.join(self.test_output_dir, test_filename)
98         if os.path.exists(test_file):
99             os.remove(test_file)
100         
101         call = self.base_call.format(self.input_file, self.test_output_dir)
102         proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True)
103         proc.wait()
104
105         copyfile(self.call_output, test_file)
106
107         baseline_file = os.path.join(".", self.baseline_output_dir, test_filename)
108
109         test = pd.read_table(test_file)
110         baseline = pd.read_table(baseline_file)
111         assert_frame_equal(test,baseline)
112
113
114     def test_collapse_user(self):
115         test_filename =  "collapse-user_" + self.wikiq_out_name
116         test_file = os.path.join(self.test_output_dir, test_filename)
117         if os.path.exists(test_file):
118             os.remove(test_file)
119         
120         call = self.base_call.format(self.input_file, self.test_output_dir)
121         call = call + " --collapse-user"
122
123         proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True)
124         proc.wait()
125
126         copyfile(self.call_output, test_file)
127
128         baseline_file = os.path.join(".", self.baseline_output_dir, test_filename)
129         test = pd.read_table(test_file)
130         baseline = pd.read_table(baseline_file)
131         assert_frame_equal(test,baseline)
132
133     def test_pwr_segment(self):
134         test_filename =  "persistence_segment_" + self.wikiq_out_name
135         test_file = os.path.join(self.test_output_dir, test_filename)
136         if os.path.exists(test_file):
137             os.remove(test_file)
138         
139         call = self.base_call.format(self.input_file, self.test_output_dir)
140         call = call + " --persistence segment"
141         proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True)
142         proc.wait()
143
144
145         copyfile(self.call_output, test_file)
146
147         baseline_file = os.path.join(".", self.baseline_output_dir, test_filename)
148
149         test = pd.read_table(test_file)
150         baseline = pd.read_table(baseline_file)
151         assert_frame_equal(test,baseline)
152
153     def test_pwr_legacy(self):
154         test_filename =  "persistence_legacy_" + self.wikiq_out_name
155         test_file = os.path.join(self.test_output_dir, test_filename)
156         if os.path.exists(test_file):
157             os.remove(test_file)
158         
159         call = self.base_call.format(self.input_file, self.test_output_dir)
160         call = call + " --persistence legacy"
161         proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True)
162         proc.wait()
163
164
165         copyfile(self.call_output, test_file)
166
167         baseline_file = os.path.join(".", self.baseline_output_dir, test_filename)
168
169         test = pd.read_table(test_file)
170         baseline = pd.read_table(baseline_file)
171         assert_frame_equal(test,baseline)
172
173     def test_pwr(self):
174         test_filename =  "persistence_" + self.wikiq_out_name
175         test_file = os.path.join(self.test_output_dir, test_filename)
176         if os.path.exists(test_file): 
177            os.remove(test_file)
178         
179         call = self.base_call.format(self.input_file, self.test_output_dir)
180         call = call + " --persistence"
181         proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True)
182         proc.wait()
183
184
185         copyfile(self.call_output, test_file)
186
187         baseline_file = os.path.join(".", self.baseline_output_dir, test_filename)
188
189         test = pd.read_table(test_file)
190         baseline = pd.read_table(baseline_file)
191         assert_frame_equal(test,baseline)
192
193
194     def test_url_encode(self):
195         test_filename =  "url-encode_" + self.wikiq_out_name
196
197         test_file = os.path.join(self.test_output_dir, test_filename)
198         if os.path.exists(test_file):
199             os.remove(test_file)
200         
201         call = self.base_call.format(self.input_file, self.test_output_dir)
202         call = call + " --url-encode"
203         proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True)
204         proc.wait()
205
206         copyfile(self.call_output, test_file)
207         baseline_file = os.path.join(".", self.baseline_output_dir, test_filename)
208         test = pd.read_table(test_file)
209         baseline = pd.read_table(baseline_file)
210         assert_frame_equal(test,baseline)
211
212
213 class Test_Malformed(unittest.TestCase):
214     def setUp(self):
215         if not os.path.exists("test_output"):
216             os.mkdir("test_output")
217
218         self.wiki = 'twinpeaks'
219         self.wikiq_out_name =  self.wiki + ".tsv"
220         self.test_output_dir = os.path.join(".", "test_output")
221         self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name)
222
223         self.infile = "{0}.xml.7z".format(self.wiki)
224         self.base_call = "../wikiq {0} -o {1}"
225         self.input_dir = "dumps"
226         self.input_file = os.path.join(".", self.input_dir,self.infile)
227
228
229     def test_malformed_noargs(self):
230
231         call = self.base_call.format(self.input_file, self.test_output_dir)
232         proc = subprocess.Popen(call,stdout=subprocess.PIPE,stderr=subprocess.PIPE, shell=True)
233         proc.wait()
234         outs, errs = proc.communicate()
235         errlines = str(errs).split("\\n")
236         self.assertEqual(errlines[-2],'xml.etree.ElementTree.ParseError: no element found: line 1369, column 0')
237
238 class Test_Stdout(unittest.TestCase):
239
240     def setUp(self):
241         self.wiki = 'sailormoon'
242         self.wikiq_out_name =  self.wiki + ".tsv"
243
244         self.infile = "{0}.xml.7z".format(self.wiki)
245         self.base_call = "../wikiq {0} --stdout"
246         self.input_dir = "dumps"
247         self.input_file = os.path.join(".", self.input_dir,self.infile)
248         self.baseline_output_dir = "baseline_output"
249
250     def test_noargs(self):
251
252         call = self.base_call.format(self.input_file)
253         proc = subprocess.run(call,stdout=subprocess.PIPE,shell=True)
254         outs = proc.stdout.decode("utf8")
255
256         test_file = "noargs_" + self.wikiq_out_name
257         baseline_file = os.path.join(".", self.baseline_output_dir, test_file)
258         print(baseline_file)
259         test = pd.read_table(StringIO(outs))
260         baseline = pd.read_table(baseline_file)
261         assert_frame_equal(test,baseline)
262         
263 if __name__ == '__main__':
264     unittest.main()

Community Data Science Collective || Want to submit a patch?