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

Standardize calling for wikiq in tests

Browse Source 
This way failures show the output of stderr/etc.

Also create path constant strings for use in tests to avoid repetition
and make changes easier.

Signed-off-by: Will Beason <willbeason@gmail.com>
This commit is contained in:
Will Beason 2025-05-27 14:27:49 -05:00
parent ebc57864f2
commit 4d3900b541
1 changed files with 50 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

103
test/Wikiq_Unit_Test.py
View File

@ -13,6 +13,27 @@ from typing import Final
TEST_DIR: Final[str] = os.path.dirname(os.path.realpath(__file__)) TEST_DIR: Final[str] = os.path.dirname(os.path.realpath(__file__))
WIKIQ: Final[str] = os.path.join(os.path.dirname(TEST_DIR), "wikiq") WIKIQ: Final[str] = os.path.join(os.path.dirname(TEST_DIR), "wikiq")
TEST_OUTPUT_DIR: Final[str] = os.path.join(TEST_DIR, "test_output") TEST_OUTPUT_DIR: Final[str] = os.path.join(TEST_DIR, "test_output")
BASELINE_OUTPUT_DIR: Final[str] = os.path.join(TEST_DIR, "baseline_output")
def setup():
tracemalloc.start()
# 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_DIR):
os.mkdir(TEST_OUTPUT_DIR)
else:
# Avoid subsequent calls to tests interfering with each other.
# Otherwise, a test may erroneously pass if the program has no output
# but a previous run output what was expected.
for f in os.listdir(TEST_OUTPUT_DIR):
os.remove(os.path.join(TEST_OUTPUT_DIR, f))
setup()
def call_wikiq(*args: str):
call = ' '.join([WIKIQ, *args])
print(call)
subprocess.check_output(call, stderr=subprocess.PIPE, shell=True)
# with / without pwr DONE # with / without pwr DONE
# with / without url encode DONE # with / without url encode DONE
@ -26,16 +47,14 @@ TEST_OUTPUT_DIR: Final[str] = os.path.join(TEST_DIR, "test_output")
class Test_Wikipedia(unittest.TestCase): class Test_Wikipedia(unittest.TestCase):
def setUp(self): def setUp(self):
self.wiki = 'ikwiki-20180301-pages-meta-history' wiki = 'ikwiki-20180301-pages-meta-history'
self.wikiq_out_name = self.wiki + ".tsv" self.wikiq_out_name = wiki + ".tsv"
self.call_output = os.path.join(TEST_OUTPUT_DIR, self.wikiq_out_name) self.call_output = os.path.join(TEST_OUTPUT_DIR, self.wikiq_out_name)
self.infile = "{0}.xml.bz2".format(self.wiki) infile = "{0}.xml.bz2".format(wiki)
self.base_call = WIKIQ + " {0} -o {1}" input_dir = os.path.join(TEST_DIR, "dumps")
self.input_dir = os.path.join(TEST_DIR, "dumps") self.input_file = os.path.join(TEST_DIR, input_dir, infile)
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
@ -43,17 +62,13 @@ class Test_Wikipedia(unittest.TestCase):
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, TEST_OUTPUT_DIR)
call = call + " --url-encode"
print(call)
try: try:
subprocess.check_output(call, stderr=subprocess.PIPE, shell=True) call_wikiq(self.input_file, "-o", TEST_OUTPUT_DIR, "--url-encode")
except subprocess.CalledProcessError as exc: except subprocess.CalledProcessError as exc:
print(exc.stderr.decode("utf8")) self.fail(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(BASELINE_OUTPUT_DIR, test_filename)
# as a test let's make sure that we get equal data frames # as a test let's make sure that we get equal data frames
test = pd.read_table(test_file) test = pd.read_table(test_file)
@ -67,15 +82,14 @@ class Test_Wikipedia(unittest.TestCase):
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, TEST_OUTPUT_DIR) try:
call = call + " -n 0 -n 1" call_wikiq(self.input_file, "-o", TEST_OUTPUT_DIR,
print(call) "-n 0", "-n 1")
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: except subprocess.CalledProcessError as exc:
proc.wait() self.fail(exc.stderr.decode("utf8"))
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(BASELINE_OUTPUT_DIR, test_filename)
# as a test let's make sure that we get equal data frames # as a test let's make sure that we get equal data frames
test = pd.read_table(test_file) test = pd.read_table(test_file)
@ -91,15 +105,14 @@ class Test_Wikipedia(unittest.TestCase):
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, TEST_OUTPUT_DIR) try:
call = call + " -n 0 -n 1 -rr 1" call_wikiq(self.input_file, "-o", TEST_OUTPUT_DIR,
print(call) "-n 0", "-n 1", "-rr 1")
with subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) as proc: except subprocess.CalledProcessError as exc:
proc.wait() self.fail(exc.stderr.decode("utf8"))
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(BASELINE_OUTPUT_DIR, test_filename)
# as a test let's make sure that we get equal data frames # as a test let's make sure that we get equal data frames
test = pd.read_table(test_file) test = pd.read_table(test_file)
@ -120,7 +133,6 @@ class Test_Basic(unittest.TestCase):
self.base_call = WIKIQ + " {0} -o {1}" self.base_call = WIKIQ + " {0} -o {1}"
self.input_dir = os.path.join(TEST_DIR, "dumps") self.input_dir = os.path.join(TEST_DIR, "dumps")
self.input_file = os.path.join(TEST_DIR, self.input_dir, self.infile) 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_noargs(self): def test_noargs(self):
@ -137,7 +149,7 @@ class Test_Basic(unittest.TestCase):
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(BASELINE_OUTPUT_DIR, test_filename)
test = pd.read_table(test_file) test = pd.read_table(test_file)
baseline = pd.read_table(baseline_file) baseline = pd.read_table(baseline_file)
@ -158,7 +170,7 @@ class Test_Basic(unittest.TestCase):
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(BASELINE_OUTPUT_DIR, test_filename)
test = pd.read_table(test_file) test = pd.read_table(test_file)
baseline = pd.read_table(baseline_file) baseline = pd.read_table(baseline_file)
assert_frame_equal(test, baseline, check_like=True) assert_frame_equal(test, baseline, check_like=True)
@ -178,7 +190,7 @@ class Test_Basic(unittest.TestCase):
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(BASELINE_OUTPUT_DIR, test_filename)
test = pd.read_table(test_file) test = pd.read_table(test_file)
baseline = pd.read_table(baseline_file) baseline = pd.read_table(baseline_file)
@ -199,7 +211,7 @@ class Test_Basic(unittest.TestCase):
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(BASELINE_OUTPUT_DIR, test_filename)
test = pd.read_table(test_file) test = pd.read_table(test_file)
baseline = pd.read_table(baseline_file) baseline = pd.read_table(baseline_file)
@ -220,7 +232,7 @@ class Test_Basic(unittest.TestCase):
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(BASELINE_OUTPUT_DIR, test_filename)
test = pd.read_table(test_file) test = pd.read_table(test_file)
baseline = pd.read_table(baseline_file) baseline = pd.read_table(baseline_file)
@ -242,7 +254,7 @@ class Test_Basic(unittest.TestCase):
self.assertEqual(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(BASELINE_OUTPUT_DIR, test_filename)
test = pd.read_table(test_file) test = pd.read_table(test_file)
baseline = pd.read_table(baseline_file) baseline = pd.read_table(baseline_file)
@ -282,7 +294,6 @@ class Test_Stdout(unittest.TestCase):
self.base_call = WIKIQ + " {0} --stdout" self.base_call = WIKIQ + " {0} --stdout"
self.input_dir = os.path.join(TEST_DIR, "dumps") self.input_dir = os.path.join(TEST_DIR, "dumps")
self.input_file = os.path.join(TEST_DIR, self.input_dir, self.infile) 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_noargs(self): def test_noargs(self):
call = self.base_call.format(self.input_file) call = self.base_call.format(self.input_file)
@ -291,7 +302,7 @@ class Test_Stdout(unittest.TestCase):
outs = proc.stdout.decode("utf8") outs = proc.stdout.decode("utf8")
test_file = "noargs_" + self.wikiq_out_name test_file = "noargs_" + self.wikiq_out_name
baseline_file = os.path.join(".", self.baseline_output_dir, test_file) baseline_file = os.path.join(BASELINE_OUTPUT_DIR, test_file)
print(baseline_file) print(baseline_file)
test = pd.read_table(StringIO(outs)) test = pd.read_table(StringIO(outs))
baseline = pd.read_table(baseline_file) baseline = pd.read_table(baseline_file)
@ -312,8 +323,6 @@ class Test_Regex(unittest.TestCase):
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 = 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 = [
# label is missing # label is missing
@ -372,7 +381,7 @@ class Test_Regex(unittest.TestCase):
test = pd.read_table(test_file) test = pd.read_table(test_file)
baseline_file = os.path.join(".", self.baseline_output_dir, test_filename) baseline_file = os.path.join(BASELINE_OUTPUT_DIR, test_filename)
baseline = pd.read_table(baseline_file) baseline = pd.read_table(baseline_file)
assert_frame_equal(test, baseline, check_like=True) assert_frame_equal(test, baseline, check_like=True)
print(i) print(i)
@ -397,22 +406,10 @@ class Test_Regex(unittest.TestCase):
test = pd.read_table(test_file) test = pd.read_table(test_file)
baseline_file = os.path.join(".", self.baseline_output_dir, test_filename) baseline_file = os.path.join(BASELINE_OUTPUT_DIR, test_filename)
baseline = pd.read_table(baseline_file) baseline = pd.read_table(baseline_file)
assert_frame_equal(test, baseline, check_like=True) assert_frame_equal(test, baseline, check_like=True)
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.
if not os.path.exists(TEST_OUTPUT_DIR):
os.mkdir(TEST_OUTPUT_DIR)
else:
# Avoid subsequent calls to tests interfering with each other.
# Otherwise, a test may erroneously pass if the program has no output
# but a previous run output what was expected.
for f in os.listdir(TEST_OUTPUT_DIR):
os.remove(os.path.join(TEST_OUTPUT_DIR, f))
unittest.main() unittest.main()