Skip to content

Commit

Permalink
Fix tests to run from anywhere (#211)
Browse files Browse the repository at this point in the history
* Get project root relative to the test files and use as parent of file paths, creating single instance of path

* Add fix for tests to CHANGELOG.md

* Reformatted with Black

* Fixed Pylint issue with order of imports
  • Loading branch information
samer-hamood authored Jan 29, 2025
1 parent eaf88f9 commit 9ab8a51
Show file tree
Hide file tree
Showing 3 changed files with 74 additions and 47 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

## Release 1.6

- Fixed tests failing (to find test files) when running from the IDE or the terminal when not in the right directory
- Added Python version to GitHub Action workflow job steps and set Black to show required formatting changes
- Upgraded pre-commit hooks (pre-commit-hooks to `v5.0.0` and ruff-pre-commit to `v0.6.0`)
- Added [run-test.sh](run-tests.sh) script that runs all checks on code
Expand Down
14 changes: 9 additions & 5 deletions functional/test/test_io.py
Original file line number Diff line number Diff line change
@@ -1,19 +1,23 @@
import unittest

from pathlib import Path
from functional.io import ReusableFile, GZFile, BZ2File, XZFile, universal_write_open

project_root = Path(__file__).parent.parent.parent.absolute()


class TestUtil(unittest.TestCase):
def test_reusable_file(self):
license_file_lf = ReusableFile("LICENSE.txt")
with open("LICENSE.txt", encoding="utf8") as license_file:
file_name = f"{project_root}/LICENSE.txt"
license_file_lf = ReusableFile(file_name)
with open(file_name, encoding="utf8") as license_file:
self.assertEqual(list(license_file), list(license_file_lf))
iter_1 = iter(license_file_lf)
iter_2 = iter(license_file_lf)
self.assertEqual(list(iter_1), list(iter_2))

def test_gzip_file(self):
file_name = "functional/test/data/test.txt.gz"
file_name = f"{project_root}/functional/test/data/test.txt.gz"
expect = [
"line0\n",
"line1\n",
Expand All @@ -31,7 +35,7 @@ def test_gzip_file(self):
self.assertListEqual(expect, list(GZFile(file_name, mode="rb")))

def test_bz2_file(self):
file_name = "functional/test/data/test.txt.bz2"
file_name = f"{project_root}/functional/test/data/test.txt.bz2"
expect = [
"line0\n",
"line1\n",
Expand All @@ -49,7 +53,7 @@ def test_bz2_file(self):
self.assertListEqual(expect, list(BZ2File(file_name, mode="rb")))

def test_xz_file(self):
file_name = "functional/test/data/test.txt.xz"
file_name = f"{project_root}/functional/test/data/test.txt.xz"
expect = [
"line0\n",
"line1\n",
Expand Down
106 changes: 64 additions & 42 deletions functional/test/test_streams.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,55 +3,63 @@
import collections
import sys
import gzip
from pathlib import Path
from platform import system
import lzma
import bz2

from functional import seq, pseq
from functional.streams import Stream, ParallelStream

project_root = Path(__file__).parent.parent.parent.absolute()


class TestStreams(unittest.TestCase):
def setUp(self):
self.seq = seq
self.seq_c_disabled = Stream(disable_compression=True)

def test_open(self):
with open("LICENSE.txt", encoding="utf8") as f:
file_name = f"{project_root}/LICENSE.txt"
with open(file_name, encoding="utf8") as f:
data = f.readlines()
self.assertListEqual(data, self.seq.open("LICENSE.txt").to_list())
self.assertListEqual(data, self.seq.open(file_name).to_list())

text = "".join(data).split(",")
self.assertListEqual(
text, self.seq.open("LICENSE.txt", delimiter=",").to_list()
)
self.assertListEqual(text, self.seq.open(file_name, delimiter=",").to_list())

with self.assertRaises(ValueError):
self.seq.open("LICENSE.txt", mode="w").to_list()
self.seq.open(file_name, mode="w").to_list()

def test_open_gzip(self):
expect = ["line0\n", "line1\n", "line2"]
self.assertListEqual(
expect,
self.seq.open("functional/test/data/test.txt.gz", mode="rt").to_list(),
self.seq.open(
f"{project_root}/functional/test/data/test.txt.gz", mode="rt"
).to_list(),
)

def test_open_bz2(self):
expect = ["line0\n", "line1\n", "line2"]
self.assertListEqual(
expect,
self.seq.open("functional/test/data/test.txt.bz2", mode="rt").to_list(),
self.seq.open(
f"{project_root}/functional/test/data/test.txt.bz2", mode="rt"
).to_list(),
)

def test_open_xz(self):
expect = ["line0\n", "line1\n", "line2"]
self.assertListEqual(
expect,
self.seq.open("functional/test/data/test.txt.xz", mode="rt").to_list(),
self.seq.open(
f"{project_root}/functional/test/data/test.txt.xz", mode="rt"
).to_list(),
)

def test_disable_compression(self):
file_name = "functional/test/data/test.txt.gz"
file_name = f"{project_root}/functional/test/data/test.txt.gz"
with open(file_name, "rb") as f:
expect = f.readlines()
self.assertListEqual(
Expand Down Expand Up @@ -119,26 +127,26 @@ def test_chain(self):
self.assertEqual([1], self.seq.chain([1]).to_list())

def test_csv(self):
result = self.seq.csv("functional/test/data/test.csv").to_list()
file_name = f"{project_root}/functional/test/data/test.csv"
result = self.seq.csv(file_name).to_list()
expect = [["1", "2", "3", "4"], ["a", "b", "c", "d"]]
self.assertEqual(expect, result)
with open("functional/test/data/test.csv", "r", encoding="utf8") as csv_file:
with open(file_name, "r", encoding="utf8") as csv_file:
self.assertEqual(expect, self.seq.csv(csv_file).to_list())
with self.assertRaises(ValueError):
self.seq.csv(1)

def test_csv_dict_reader(self):
result = self.seq.csv_dict_reader(
"functional/test/data/test_header.csv"
).to_list()
file_name = f"{project_root}/functional/test/data/test_header.csv"
result = self.seq.csv_dict_reader(file_name).to_list()
self.assertEqual(result[0]["a"], "1")
self.assertEqual(result[0]["b"], "2")
self.assertEqual(result[0]["c"], "3")
self.assertEqual(result[1]["a"], "4")
self.assertEqual(result[1]["b"], "5")
self.assertEqual(result[1]["c"], "6")

with open("functional/test/data/test_header.csv", "r", encoding="utf8") as f:
with open(file_name, "r", encoding="utf8") as f:
result = self.seq.csv_dict_reader(f).to_list()
self.assertEqual(result[0]["a"], "1")
self.assertEqual(result[0]["b"], "2")
Expand All @@ -151,52 +159,66 @@ def test_csv_dict_reader(self):
self.seq.csv_dict_reader(1)

def test_gzip_csv(self):
result = self.seq.csv("functional/test/data/test.csv.gz").to_list()
result = self.seq.csv(
f"{project_root}/functional/test/data/test.csv.gz"
).to_list()
expect = [["1", "2", "3", "4"], ["a", "b", "c", "d"]]
self.assertEqual(expect, result)
with self.assertRaises(ValueError):
self.seq.csv(1)

def test_bz2_csv(self):
result = self.seq.csv("functional/test/data/test.csv.bz2").to_list()
result = self.seq.csv(
f"{project_root}/functional/test/data/test.csv.bz2"
).to_list()
expect = [["1", "2", "3", "4"], ["a", "b", "c", "d"]]
self.assertEqual(expect, result)
with self.assertRaises(ValueError):
self.seq.csv(1)

def test_xz_csv(self):
result = self.seq.csv("functional/test/data/test.csv.xz").to_list()
result = self.seq.csv(
f"{project_root}/functional/test/data/test.csv.xz"
).to_list()
expect = [["1", "2", "3", "4"], ["a", "b", "c", "d"]]
self.assertEqual(expect, result)
with self.assertRaises(ValueError):
self.seq.csv(1)

def test_jsonl(self):
result_0 = self.seq.jsonl("functional/test/data/test.jsonl").to_list()
result_0 = self.seq.jsonl(
f"{project_root}/functional/test/data/test.jsonl"
).to_list()
expect_0 = [[1, 2, 3], {"a": 1, "b": 2, "c": 3}]
self.assertEqual(expect_0, result_0)
result_1 = self.seq.jsonl(["[1, 2, 3]", "[4, 5, 6]"])
expect_1 = [[1, 2, 3], [4, 5, 6]]
self.assertEqual(expect_1, result_1)

def test_gzip_jsonl(self):
result_0 = self.seq.jsonl("functional/test/data/test.jsonl.gz").to_list()
result_0 = self.seq.jsonl(
f"{project_root}/functional/test/data/test.jsonl.gz"
).to_list()
expect_0 = [[1, 2, 3], {"a": 1, "b": 2, "c": 3}]
self.assertEqual(expect_0, result_0)

def test_bz2_jsonl(self):
result_0 = self.seq.jsonl("functional/test/data/test.jsonl.bz2").to_list()
result_0 = self.seq.jsonl(
f"{project_root}/functional/test/data/test.jsonl.bz2"
).to_list()
expect_0 = [[1, 2, 3], {"a": 1, "b": 2, "c": 3}]
self.assertEqual(expect_0, result_0)

def test_xz_jsonl(self):
result_0 = self.seq.jsonl("functional/test/data/test.jsonl.xz").to_list()
result_0 = self.seq.jsonl(
f"{project_root}/functional/test/data/test.jsonl.xz"
).to_list()
expect_0 = [[1, 2, 3], {"a": 1, "b": 2, "c": 3}]
self.assertEqual(expect_0, result_0)

def test_json(self):
list_test_path = "functional/test/data/test_list.json"
dict_test_path = "functional/test/data/test_dict.json"
list_test_path = f"{project_root}/functional/test/data/test_list.json"
dict_test_path = f"{project_root}/functional/test/data/test_dict.json"
list_expect = [1, 2, 3, 4, 5]
dict_expect = list({"a": 1, "b": 2, "c": 3}.items())

Expand All @@ -216,8 +238,8 @@ def test_json(self):
self.seq.json(1)

def test_gzip_json(self):
list_test_path = "functional/test/data/test_list.json.gz"
dict_test_path = "functional/test/data/test_dict.json.gz"
list_test_path = f"{project_root}/functional/test/data/test_list.json.gz"
dict_test_path = f"{project_root}/functional/test/data/test_dict.json.gz"
list_expect = [1, 2, 3, 4, 5]
dict_expect = list({"a": 1, "b": 2, "c": 3}.items())

Expand All @@ -230,8 +252,8 @@ def test_gzip_json(self):
self.seq.json(1)

def test_bz2_json(self):
list_test_path = "functional/test/data/test_list.json.bz2"
dict_test_path = "functional/test/data/test_dict.json.bz2"
list_test_path = f"{project_root}/functional/test/data/test_list.json.bz2"
dict_test_path = f"{project_root}/functional/test/data/test_dict.json.bz2"
list_expect = [1, 2, 3, 4, 5]
dict_expect = list({"a": 1, "b": 2, "c": 3}.items())

Expand All @@ -244,8 +266,8 @@ def test_bz2_json(self):
self.seq.json(1)

def test_xz_json(self):
list_test_path = "functional/test/data/test_list.json.xz"
dict_test_path = "functional/test/data/test_dict.json.xz"
list_test_path = f"{project_root}/functional/test/data/test_list.json.xz"
dict_test_path = f"{project_root}/functional/test/data/test_dict.json.xz"
list_expect = [1, 2, 3, 4, 5]
dict_expect = list({"a": 1, "b": 2, "c": 3}.items())

Expand All @@ -258,7 +280,7 @@ def test_xz_json(self):
self.seq.json(1)

def test_sqlite3(self):
db_file = "functional/test/data/test_sqlite3.db"
db_file = f"{project_root}/functional/test/data/test_sqlite3.db"

# test failure case
with self.assertRaises(ValueError):
Expand Down Expand Up @@ -313,7 +335,7 @@ def test_pandas(self):
pass

def test_to_file(self):
tmp_path = "functional/test/data/tmp/output.txt"
tmp_path = f"{project_root}/functional/test/data/tmp/output.txt"
sequence = self.seq(1, 2, 3, 4)
sequence.to_file(tmp_path)
with open(tmp_path, "r", encoding="utf8") as output:
Expand All @@ -324,7 +346,7 @@ def test_to_file(self):
self.assertEqual("1:2:3:4", output.readlines()[0])

def test_to_file_compressed(self):
tmp_path = "functional/test/data/tmp/output.txt"
tmp_path = f"{project_root}/functional/test/data/tmp/output.txt"
sequence = self.seq(1, 2, 3, 4)
sequence.to_file(tmp_path, compression="gzip")
with gzip.open(tmp_path, "rt") as output:
Expand All @@ -339,7 +361,7 @@ def test_to_file_compressed(self):
self.assertEqual("[1, 2, 3, 4]", output.readlines()[0])

def test_to_jsonl(self):
tmp_path = "functional/test/data/tmp/output.txt"
tmp_path = f"{project_root}/functional/test/data/tmp/output.txt"
elements = [{"a": 1, "b": 2}, {"c": 3}, {"d": 4}]
sequence = self.seq(elements)

Expand All @@ -348,7 +370,7 @@ def test_to_jsonl(self):
self.assertEqual(elements, result)

def test_to_jsonl_compressed(self):
tmp_path = "functional/test/data/tmp/output.txt"
tmp_path = f"{project_root}/functional/test/data/tmp/output.txt"
elements = [{"a": 1, "b": 2}, {"c": 3}, {"d": 4}]
sequence = self.seq(elements)

Expand All @@ -365,7 +387,7 @@ def test_to_jsonl_compressed(self):
self.assertEqual(elements, result)

def test_to_json(self):
tmp_path = "functional/test/data/tmp/output.txt"
tmp_path = f"{project_root}/functional/test/data/tmp/output.txt"
elements = [["a", 1], ["b", 2], ["c", 3]]
sequence = self.seq(elements)

Expand All @@ -379,7 +401,7 @@ def test_to_json(self):
self.assertEqual(dict_expect, result)

def test_to_json_compressed(self):
tmp_path = "functional/test/data/tmp/output.txt"
tmp_path = f"{project_root}/functional/test/data/tmp/output.txt"
elements = [["a", 1], ["b", 2], ["c", 3]]
dict_expect = {"a": 1, "b": 2, "c": 3}
sequence = self.seq(elements)
Expand Down Expand Up @@ -409,7 +431,7 @@ def test_to_json_compressed(self):
self.assertEqual(dict_expect, result)

def test_to_csv(self):
tmp_path = "functional/test/data/tmp/output.txt"
tmp_path = f"{project_root}/functional/test/data/tmp/output.txt"
elements = [[1, 2, 3], [4, 5, 6], ["a", "b", "c"]]
expect = [["1", "2", "3"], ["4", "5", "6"], ["a", "b", "c"]]
sequence = self.seq(elements)
Expand All @@ -419,7 +441,7 @@ def test_to_csv(self):

@unittest.skipUnless(system().startswith("Win"), "Skip CSV test if not on Windows")
def test_to_csv_win(self):
tmp_path = "functional/test/data/tmp/output.txt"
tmp_path = f"{project_root}/functional/test/data/tmp/output.txt"
elements = [[1, 2, 3], [4, 5, 6], ["a", "b", "c"]]
expect = [["1", "2", "3"], [], ["4", "5", "6"], [], ["a", "b", "c"], []]
sequence = self.seq(elements)
Expand All @@ -428,7 +450,7 @@ def test_to_csv_win(self):
self.assertNotEqual(expect, result)

def test_to_csv_compressed(self):
tmp_path = "functional/test/data/tmp/output.txt"
tmp_path = f"{project_root}/functional/test/data/tmp/output.txt"
elements = [[1, 2, 3], [4, 5, 6], ["a", "b", "c"]]
expect = [["1", "2", "3"], ["4", "5", "6"], ["a", "b", "c"]]
sequence = self.seq(elements)
Expand All @@ -452,7 +474,7 @@ def test_to_sqlite3_failure(self):
self.seq(elements).to_sqlite3(1, insert_sql)

def test_to_sqlite3_file(self):
tmp_path = "functional/test/data/tmp/test.db"
tmp_path = f"{project_root}/functional/test/data/tmp/test.db"

with sqlite3.connect(tmp_path) as conn:
conn.execute("DROP TABLE IF EXISTS user;")
Expand Down

0 comments on commit 9ab8a51

Please sign in to comment.