collective/mediawiki_dump_tools
16
0
Fork 0
Code Issues Pull Requests Activity

Make tests runnable from anywhere

Browse Source 
Tests no longer implicitly require that the caller be in
a specific working directory.

Signed-off-by: Will Beason <willbeason@gmail.com>
This commit is contained in:
Will Beason 2025-05-27 13:40:57 -05:00
parent 3d0bf89938
commit ebc57864f2
3 changed files with 79 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
README.rst
View File

@ -20,6 +20,11 @@ associated tests to work.
- 7zip - 7zip
- ffmpeg - ffmpeg
Tests
----
To run tests::
python -m unittest test.Wikiq_Unit_Test
TODO: TODO:
_______________ _______________

150
test/Wikiq_Unit_Test.py
View File

@ -6,9 +6,13 @@ import pandas as pd
from pandas.testing import assert_frame_equal from pandas.testing import assert_frame_equal
from io import StringIO from io import StringIO
import tracemalloc import tracemalloc
from typing import Final
tracemalloc.start()
# Make references to files and wikiq relative to this file, not to the current working directory.
TEST_DIR: Final[str] = os.path.dirname(os.path.realpath(__file__))
WIKIQ: Final[str] = os.path.join(os.path.dirname(TEST_DIR), "wikiq")
TEST_OUTPUT_DIR: Final[str] = os.path.join(TEST_DIR, "test_output")
# with / without pwr DONE # with / without pwr DONE
# with / without url encode DONE # with / without url encode DONE
@ -24,27 +28,29 @@ class Test_Wikipedia(unittest.TestCase):
def setUp(self): def setUp(self):
self.wiki = 'ikwiki-20180301-pages-meta-history' self.wiki = 'ikwiki-20180301-pages-meta-history'
self.wikiq_out_name = self.wiki + ".tsv" self.wikiq_out_name = self.wiki + ".tsv"
self.test_output_dir = os.path.join(".", "test_output") self.call_output = os.path.join(TEST_OUTPUT_DIR, self.wikiq_out_name)
self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name)
self.infile = "{0}.xml.bz2".format(self.wiki) self.infile = "{0}.xml.bz2".format(self.wiki)
self.base_call = "../wikiq {0} -o {1}"
self.input_dir = "dumps" self.base_call = WIKIQ + " {0} -o {1}"
self.input_file = os.path.join(".", self.input_dir, self.infile) self.input_dir = os.path.join(TEST_DIR, "dumps")
self.baseline_output_dir = "baseline_output" self.input_file = os.path.join(TEST_DIR, self.input_dir, self.infile)
self.baseline_output_dir = os.path.join(TEST_DIR, "baseline_output")
def test_WP_url_encode(self): def test_WP_url_encode(self):
test_filename = "url-encode_" + self.wikiq_out_name test_filename = "url-encode_" + self.wikiq_out_name
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " --url-encode" call = call + " --url-encode"
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: try:
proc.wait() subprocess.check_output(call, stderr=subprocess.PIPE, shell=True)
assert (proc.returncode == 0) except subprocess.CalledProcessError as exc:
print(exc.stderr.decode("utf8"))
self.fail()
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
baseline_file = os.path.join(".", self.baseline_output_dir, test_filename) baseline_file = os.path.join(".", self.baseline_output_dir, test_filename)
@ -57,16 +63,16 @@ class Test_Wikipedia(unittest.TestCase):
def test_WP_namespaces(self): def test_WP_namespaces(self):
print(os.path.abspath('.')) print(os.path.abspath('.'))
test_filename = "namespaces_" + self.wikiq_out_name test_filename = "namespaces_" + self.wikiq_out_name
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " -n 0 -n 1" call = call + " -n 0 -n 1"
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
baseline_file = os.path.join(os.path.abspath("."), self.baseline_output_dir, test_filename) baseline_file = os.path.join(os.path.abspath("."), self.baseline_output_dir, test_filename)
@ -81,16 +87,16 @@ class Test_Wikipedia(unittest.TestCase):
def test_WP_revert_radius(self): def test_WP_revert_radius(self):
print(os.path.abspath('.')) print(os.path.abspath('.'))
test_filename = "revert_radius_" + self.wikiq_out_name test_filename = "revert_radius_" + self.wikiq_out_name
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " -n 0 -n 1 -rr 1" call = call + " -n 0 -n 1 -rr 1"
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
baseline_file = os.path.join(os.path.abspath("."), self.baseline_output_dir, test_filename) baseline_file = os.path.join(os.path.abspath("."), self.baseline_output_dir, test_filename)
@ -108,27 +114,26 @@ class Test_Basic(unittest.TestCase):
def setUp(self): def setUp(self):
self.wiki = 'sailormoon' self.wiki = 'sailormoon'
self.wikiq_out_name = self.wiki + ".tsv" self.wikiq_out_name = self.wiki + ".tsv"
self.test_output_dir = os.path.join(".", "test_output") self.call_output = os.path.join(TEST_OUTPUT_DIR, self.wikiq_out_name)
self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name)
self.infile = "{0}.xml.7z".format(self.wiki) self.infile = "{0}.xml.7z".format(self.wiki)
self.base_call = "../wikiq {0} -o {1}" self.base_call = WIKIQ + " {0} -o {1}"
self.input_dir = "dumps" self.input_dir = os.path.join(TEST_DIR, "dumps")
self.input_file = os.path.join(".", self.input_dir, self.infile) self.input_file = os.path.join(TEST_DIR, self.input_dir, self.infile)
self.baseline_output_dir = "baseline_output" self.baseline_output_dir = os.path.join(TEST_DIR, "baseline_output")
def test_noargs(self): def test_noargs(self):
test_filename = "noargs_" + self.wikiq_out_name test_filename = "noargs_" + self.wikiq_out_name
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
@ -140,16 +145,16 @@ class Test_Basic(unittest.TestCase):
def test_collapse_user(self): def test_collapse_user(self):
test_filename = "collapse-user_" + self.wikiq_out_name test_filename = "collapse-user_" + self.wikiq_out_name
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " --collapse-user" call = call + " --collapse-user"
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
@ -160,16 +165,16 @@ class Test_Basic(unittest.TestCase):
def test_pwr_segment(self): def test_pwr_segment(self):
test_filename = "persistence_segment_" + self.wikiq_out_name test_filename = "persistence_segment_" + self.wikiq_out_name
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " --persistence segment" call = call + " --persistence segment"
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
@ -181,16 +186,16 @@ class Test_Basic(unittest.TestCase):
def test_pwr_legacy(self): def test_pwr_legacy(self):
test_filename = "persistence_legacy_" + self.wikiq_out_name test_filename = "persistence_legacy_" + self.wikiq_out_name
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " --persistence legacy" call = call + " --persistence legacy"
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
@ -202,16 +207,16 @@ class Test_Basic(unittest.TestCase):
def test_pwr(self): def test_pwr(self):
test_filename = "persistence_" + self.wikiq_out_name test_filename = "persistence_" + self.wikiq_out_name
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " --persistence" call = call + " --persistence"
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
@ -226,15 +231,15 @@ class Test_Basic(unittest.TestCase):
def test_url_encode(self): def test_url_encode(self):
test_filename = "url-encode_" + self.wikiq_out_name test_filename = "url-encode_" + self.wikiq_out_name
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " --url-encode" call = call + " --url-encode"
with subprocess.Popen(call, stdout=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
baseline_file = os.path.join(".", self.baseline_output_dir, test_filename) baseline_file = os.path.join(".", self.baseline_output_dir, test_filename)
@ -247,21 +252,17 @@ class Test_Basic(unittest.TestCase):
class Test_Malformed(unittest.TestCase): class Test_Malformed(unittest.TestCase):
def setUp(self): def setUp(self):
if not os.path.exists("test_output"):
os.mkdir("test_output")
self.wiki = 'twinpeaks' self.wiki = 'twinpeaks'
self.wikiq_out_name = self.wiki + ".tsv" self.wikiq_out_name = self.wiki + ".tsv"
self.test_output_dir = os.path.join(".", "test_output") self.call_output = os.path.join(TEST_OUTPUT_DIR, self.wikiq_out_name)
self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name)
self.infile = "{0}.xml.7z".format(self.wiki) self.infile = "{0}.xml.7z".format(self.wiki)
self.base_call = "../wikiq {0} -o {1}" self.base_call = WIKIQ + " {0} -o {1}"
self.input_dir = "dumps" self.input_dir = os.path.join(TEST_DIR, "dumps")
self.input_file = os.path.join(".", self.input_dir, self.infile) self.input_file = os.path.join(TEST_DIR, self.input_dir, self.infile)
def test_malformed_noargs(self): def test_malformed_noargs(self):
call = self.base_call.format(self.input_file, self.test_output_dir) call = self.base_call.format(self.input_file, TEST_OUTPUT_DIR)
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
@ -278,10 +279,10 @@ class Test_Stdout(unittest.TestCase):
self.wikiq_out_name = self.wiki + ".tsv" self.wikiq_out_name = self.wiki + ".tsv"
self.infile = "{0}.xml.7z".format(self.wiki) self.infile = "{0}.xml.7z".format(self.wiki)
self.base_call = "../wikiq {0} --stdout" self.base_call = WIKIQ + " {0} --stdout"
self.input_dir = "dumps" self.input_dir = os.path.join(TEST_DIR, "dumps")
self.input_file = os.path.join(".", self.input_dir, self.infile) self.input_file = os.path.join(TEST_DIR, self.input_dir, self.infile)
self.baseline_output_dir = "baseline_output" self.baseline_output_dir = os.path.join(TEST_DIR, "baseline_output")
def test_noargs(self): def test_noargs(self):
call = self.base_call.format(self.input_file) call = self.base_call.format(self.input_file)
@ -298,25 +299,20 @@ class Test_Stdout(unittest.TestCase):
class Test_Regex(unittest.TestCase): class Test_Regex(unittest.TestCase):
def setUp(self): def setUp(self):
self.wiki = 'regextest' self.wiki = 'regextest'
self.wikiq_out_name = self.wiki + '.tsv' self.wikiq_out_name = self.wiki + '.tsv'
self.infile = "{0}.xml.bz2".format(self.wiki) self.infile = "{0}.xml.bz2".format(self.wiki)
self.input_dir = "dumps" self.input_dir = os.path.join(TEST_DIR, "dumps")
self.input_file = os.path.join(".", self.input_dir, self.infile) self.input_file = os.path.join(TEST_DIR, self.input_dir, self.infile)
if not os.path.exists("test_output"): self.call_output = os.path.join(TEST_OUTPUT_DIR, self.wikiq_out_name)
os.mkdir("test_output")
self.test_output_dir = os.path.join(".", "test_output")
self.call_output = os.path.join(self.test_output_dir, self.wikiq_out_name)
# we have two base calls, one for checking arguments and the other for checking outputs # we have two base calls, one for checking arguments and the other for checking outputs
self.base_call = "../wikiq {0}" self.base_call = WIKIQ + " {0}"
self.base_call_outs = "../wikiq {0} -o {1}" self.base_call_outs = WIKIQ + " {0} -o {1}"
self.baseline_output_dir = "baseline_output" self.baseline_output_dir = os.path.join(TEST_DIR, "baseline_output")
# sample arguments for checking that bad arguments get terminated / test_regex_arguments # sample arguments for checking that bad arguments get terminated / test_regex_arguments
self.bad_arguments_list = [ self.bad_arguments_list = [
@ -361,16 +357,16 @@ class Test_Regex(unittest.TestCase):
test_filename = "basic_{0}_{1}.tsv".format(self.wikiq_out_name[:-4], str(i)) test_filename = "basic_{0}_{1}.tsv".format(self.wikiq_out_name[:-4], str(i))
# print(test_filename) # print(test_filename)
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call_outs.format(self.input_file, self.test_output_dir) call = self.base_call_outs.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " " + arguments call = call + " " + arguments
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
@ -385,17 +381,17 @@ class Test_Regex(unittest.TestCase):
for i, arguments in enumerate(self.cap_arguments_list): for i, arguments in enumerate(self.cap_arguments_list):
test_filename = "capturegroup_{0}_{1}.tsv".format(self.wikiq_out_name[:-4], str(i)) test_filename = "capturegroup_{0}_{1}.tsv".format(self.wikiq_out_name[:-4], str(i))
print(test_filename) print(test_filename)
test_file = os.path.join(self.test_output_dir, test_filename) test_file = os.path.join(TEST_OUTPUT_DIR, test_filename)
if os.path.exists(test_file): if os.path.exists(test_file):
os.remove(test_file) os.remove(test_file)
call = self.base_call_outs.format(self.input_file, self.test_output_dir) call = self.base_call_outs.format(self.input_file, TEST_OUTPUT_DIR)
call = call + " " + arguments call = call + " " + arguments
print(call) print(call)
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc:
proc.wait() proc.wait()
assert (proc.returncode == 0) self.assertEqual(proc.returncode, 0)
copyfile(self.call_output, test_file) copyfile(self.call_output, test_file)
@ -407,14 +403,16 @@ class Test_Regex(unittest.TestCase):
if __name__ == '__main__': if __name__ == '__main__':
tracemalloc.start()
# Perform directory check and reset here as this is a one-time setup step as opposed to per-test setup. # Perform directory check and reset here as this is a one-time setup step as opposed to per-test setup.
if not os.path.exists("test_output"): if not os.path.exists(TEST_OUTPUT_DIR):
os.mkdir("test_output") os.mkdir(TEST_OUTPUT_DIR)
else: else:
# Avoid subsequent calls to tests interfering with each other. # Avoid subsequent calls to tests interfering with each other.
# Otherwise, a test may erroneously pass if the program has no output # Otherwise, a test may erroneously pass if the program has no output
# but a previous run output what was expected. # but a previous run output what was expected.
for f in os.listdir("test_output"): for f in os.listdir(TEST_OUTPUT_DIR):
os.remove(os.path.join("test_output", f)) os.remove(os.path.join(TEST_OUTPUT_DIR, f))
unittest.main() unittest.main()

0
test/__init__.py Normal file
View File