X-Git-Url: https://code.communitydata.science/mediawiki_dump_tools.git/blobdiff_plain/f69e8b44a67214d8aa424fff33e116f2e1bf567a..3d12865c4ec7db67123443abb3d5dc26ee0f300d:/tests/Wikiq_Test.py diff --git a/tests/Wikiq_Test.py b/tests/Wikiq_Test.py index e43ee7d..09cb3c5 100644 --- a/tests/Wikiq_Test.py +++ b/tests/Wikiq_Test.py @@ -16,30 +16,26 @@ from io import StringIO # wikia and wikipedia data DONE # malformed xmls DONE -class Test_Wikipedia(unittest.TestCase): - def setUp(self): +class Test_Wikiq(unittest.TestCase): + + def mkoutputdir(self): if not os.path.exists("test_output"): os.mkdir("test_output") - self.wiki = 'ikwiki-20180301-pages-meta-history' + def setuptoutputfiles(self, suffix="xml.7z"): self.wikiq_out_name = self.wiki + ".tsv" self.test_output_dir = os.path.join(".", "test_output") self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name) - - self.infile = "{0}.xml.bz2".format(self.wiki) - self.base_call = "../wikiq {0} -o {1}" + self.infile = "{0}.{1}".format(self.wiki,suffix) self.input_dir = "dumps" self.input_file = os.path.join(".", self.input_dir,self.infile) self.baseline_output_dir = "baseline_output" - def test_WP_url_encode(self): - test_filename = "url-encode_" + self.wikiq_out_name + def run_and_check_output(self, call, test_filename): test_file = os.path.join(self.test_output_dir, test_filename) if os.path.exists(test_file): os.remove(test_file) - - call = self.base_call.format(self.input_file, self.test_output_dir) - call = call + " --url-encode" + proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True) proc.wait() @@ -51,142 +47,74 @@ class Test_Wikipedia(unittest.TestCase): baseline = pd.read_table(baseline_file) assert_frame_equal(test,baseline) +class Test_Wikipedia(Test_Wikiq): + def setUp(self): + print(os.path.abspath(".")) + self.mkoutputdir() + self.wiki = 'ikwiki-20180301-pages-meta-history' + self.setuptoutputfiles(suffix="xml.bz2") + self.base_call = "../bin/wikiq {0} -o {1}" -class Test_Basic(unittest.TestCase): + def test_WP_url_encode(self): + test_filename = "url-encode_" + self.wikiq_out_name + call = self.base_call.format(self.input_file, self.test_output_dir) + call = call + " --url-encode" + self.run_and_check_output(call, test_filename) - def setUp(self): - if not os.path.exists("test_output"): - os.mkdir("test_output") - self.wiki = 'sailormoon' - self.wikiq_out_name = self.wiki + ".tsv" - self.test_output_dir = os.path.join(".", "test_output") - self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name) +class Test_Basic(Test_Wikiq): - self.infile = "{0}.xml.7z".format(self.wiki) - self.base_call = "../wikiq {0} -o {1}" - self.input_dir = "dumps" - self.input_file = os.path.join(".", self.input_dir,self.infile) - self.baseline_output_dir = "baseline_output" + def setUp(self): + self.mkoutputdir() + self.wiki="sailormoon" + self.setuptoutputfiles() + self.base_call = "../bin/wikiq {0} -o {1}" def test_noargs(self): - test_filename = "noargs_" + self.wikiq_out_name - test_file = os.path.join(self.test_output_dir, test_filename) - if os.path.exists(test_file): - os.remove(test_file) call = self.base_call.format(self.input_file, self.test_output_dir) - proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True) - proc.wait() - - copyfile(self.call_output, test_file) - - baseline_file = os.path.join(".", self.baseline_output_dir, test_filename) - - test = pd.read_table(test_file) - baseline = pd.read_table(baseline_file) - assert_frame_equal(test,baseline) - + print(call) + self.run_and_check_output(call, test_filename) def test_collapse_user(self): test_filename = "collapse-user_" + self.wikiq_out_name - test_file = os.path.join(self.test_output_dir, test_filename) - if os.path.exists(test_file): - os.remove(test_file) call = self.base_call.format(self.input_file, self.test_output_dir) call = call + " --collapse-user" - proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True) - proc.wait() - - copyfile(self.call_output, test_file) - - baseline_file = os.path.join(".", self.baseline_output_dir, test_filename) - test = pd.read_table(test_file) - baseline = pd.read_table(baseline_file) - assert_frame_equal(test,baseline) + self.run_and_check_output(call, test_filename) def test_pwr_legacy(self): test_filename = "persistence_legacy_" + self.wikiq_out_name - test_file = os.path.join(self.test_output_dir, test_filename) - if os.path.exists(test_file): - os.remove(test_file) call = self.base_call.format(self.input_file, self.test_output_dir) call = call + " --persistence-legacy" - proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True) - proc.wait() - - - copyfile(self.call_output, test_file) - - baseline_file = os.path.join(".", self.baseline_output_dir, test_filename) - - test = pd.read_table(test_file) - baseline = pd.read_table(baseline_file) - assert_frame_equal(test,baseline) + self.run_and_check_output(call, test_filename) def test_pwr(self): test_filename = "persistence_" + self.wikiq_out_name - test_file = os.path.join(self.test_output_dir, test_filename) - if os.path.exists(test_file): - os.remove(test_file) call = self.base_call.format(self.input_file, self.test_output_dir) call = call + " --persistence" - proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True) - proc.wait() - - - copyfile(self.call_output, test_file) - - baseline_file = os.path.join(".", self.baseline_output_dir, test_filename) - - test = pd.read_table(test_file) - baseline = pd.read_table(baseline_file) - assert_frame_equal(test,baseline) - + self.run_and_check_output(call, test_filename) def test_url_encode(self): test_filename = "url-encode_" + self.wikiq_out_name - test_file = os.path.join(self.test_output_dir, test_filename) - if os.path.exists(test_file): - os.remove(test_file) - call = self.base_call.format(self.input_file, self.test_output_dir) call = call + " --url-encode" - proc = subprocess.Popen(call,stdout=subprocess.PIPE,shell=True) - proc.wait() - - copyfile(self.call_output, test_file) - baseline_file = os.path.join(".", self.baseline_output_dir, test_filename) - test = pd.read_table(test_file) - baseline = pd.read_table(baseline_file) - assert_frame_equal(test,baseline) - + self.run_and_check_output(call, test_filename) -class Test_Malformed(unittest.TestCase): +class Test_Malformed(Test_Wikiq): def setUp(self): - if not os.path.exists("test_output"): - os.mkdir("test_output") - - self.wiki = 'twinpeaks' - self.wikiq_out_name = self.wiki + ".tsv" - self.test_output_dir = os.path.join(".", "test_output") - self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name) - - self.infile = "{0}.xml.7z".format(self.wiki) - self.base_call = "../wikiq {0} -o {1}" - self.input_dir = "dumps" - self.input_file = os.path.join(".", self.input_dir,self.infile) - + self.mkoutputdir() + self.wiki="twinpeaks" + self.setuptoutputfiles() + self.base_call = "../bin/wikiq {0} -o {1}" def test_malformed_noargs(self): - call = self.base_call.format(self.input_file, self.test_output_dir) proc = subprocess.Popen(call,stdout=subprocess.PIPE,stderr=subprocess.PIPE, shell=True) proc.wait() @@ -194,28 +122,21 @@ class Test_Malformed(unittest.TestCase): errlines = str(errs).split("\\n") self.assertEqual(errlines[-2],'xml.etree.ElementTree.ParseError: no element found: line 1369, column 0') -class Test_Stdout(unittest.TestCase): +class Test_Stdout(Test_Wikiq): def setUp(self): + self.mkoutputdir() self.wiki = 'sailormoon' - self.wikiq_out_name = self.wiki + ".tsv" + self.setuptoutputfiles() - self.infile = "{0}.xml.7z".format(self.wiki) - self.base_call = "../wikiq {0} --stdout" - self.input_dir = "dumps" - self.input_file = os.path.join(".", self.input_dir,self.infile) - self.baseline_output_dir = "baseline_output" def test_noargs(self): - - call = self.base_call.format(self.input_file) - proc = subprocess.run(call,stdout=subprocess.PIPE,shell=True) - outs = proc.stdout.decode("utf8") - + self.base_call = ["../bin/wikiq", self.input_file, "--stdout"] + proc = subprocess.Popen(self.base_call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8') + outs = proc.stdout test_file = "noargs_" + self.wikiq_out_name baseline_file = os.path.join(".", self.baseline_output_dir, test_file) - print(baseline_file) - test = pd.read_table(StringIO(outs)) + test = pd.read_table(outs) baseline = pd.read_table(baseline_file) assert_frame_equal(test,baseline)