Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix tests to run from anywhere #211

Merged
merged 7 commits into from
Jan 29, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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