-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_data_processor.py
More file actions
133 lines (104 loc) · 4.81 KB
/
Copy pathtest_data_processor.py
File metadata and controls
133 lines (104 loc) · 4.81 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
import os
import unittest
import pandas as pd
import numpy as np
from data_processor import DataProcessor
class DataProcessorTest(unittest.TestCase):
def test_constructor(self):
processor = DataProcessor(filepath="sample_dataset.csv")
self.assertIsNotNone(processor)
self.assertIsInstance(processor.df, pd.DataFrame)
def test_constructor_raise_error(self):
filepath = "NON_EXISTENT_FILE.csv"
with self.assertRaises(Exception):
processor = DataProcessor(filepath)
def test_constructor_filter_by_date(self):
filepath = "sample_dataset.csv"
start_date = "2023-01-16"
end_date = "2023-02-20"
processor = DataProcessor(
filepath,
start_date,
end_date,
)
filter_by_date = processor._filter_by_date
self.assertIsNotNone(processor)
self.assertIsInstance(processor.df, pd.DataFrame)
# test_start_date : 2023-01-20
# test_end_date : 2023-02-18
test_start_date = processor.df["Date"].iloc[0]
test_end_date = processor.df["Date"].iloc[-1]
self.assertGreaterEqual(test_start_date, filter_by_date[0])
self.assertLessEqual(test_end_date, filter_by_date[1])
def test_constructor_filter_by_date_raise_error(self):
from pandas._libs.tslibs.parsing import DateParseError
with self.assertRaises(DateParseError):
processor = DataProcessor(
filepath="sample_dataset.csv",
start_date="2023-01-a0", # day can not be 0
end_date="2023-02-20",
)
def test_constructor_filter_by_region(self):
processor_n = DataProcessor(filepath="sample_dataset.csv", region="North")
processor_e = DataProcessor(filepath="sample_dataset.csv", region="East")
processor_s = DataProcessor(filepath="sample_dataset.csv", region="South")
processor_w = DataProcessor(filepath="sample_dataset.csv", region="West")
self.assertIsNotNone(processor_n)
self.assertIsInstance(processor_n.df, pd.DataFrame)
self.assertIsNotNone(processor_e)
self.assertIsInstance(processor_e.df, pd.DataFrame)
self.assertIsNotNone(processor_s)
self.assertIsInstance(processor_s.df, pd.DataFrame)
self.assertIsNotNone(processor_w)
self.assertIsInstance(processor_w.df, pd.DataFrame)
def test_constructor_filter_by_region_raise_error(self):
with self.assertRaises(KeyError):
p = DataProcessor(filepath="sample_dataset.csv", region="WRONGREGION")
# becomes an empty dataframe, keyerror an
def test_summarize_column(self):
processor = DataProcessor(filepath="sample_dataset.csv")
sales = processor.summarize_column(column="Sales")
profit = processor.summarize_column(column="Profit")
self.assertIsNotNone(sales)
self.assertIsInstance(sales, np.int64)
self.assertIsNotNone(profit)
self.assertIsInstance(profit, np.int64)
def test_summarize_column_raise_error(self):
processor = DataProcessor(filepath="sample_dataset.csv")
with self.assertRaises(KeyError):
raises_error = processor.summarize_column(column="Not a column")
def test_region_with_highest_sales(self):
processor = DataProcessor(filepath="sample_dataset.csv")
highest_sales = processor.region_with_highest_sales()
self.assertIsNotNone(highest_sales)
self.assertIsInstance(highest_sales, str)
def test_region_summary(self):
processor = DataProcessor(filepath="sample_dataset.csv")
region_summary = processor.region_summary()
self.assertIsNotNone(region_summary)
self.assertIsInstance(region_summary, pd.DataFrame)
def test_save_summary_json(self):
processor_json = DataProcessor(filepath="sample_dataset.csv")
processor_json.save_summary(file_name="test_summary", file_type="json")
self.assertTrue(os.path.exists("test_summary.json"))
def test_save_summary_csv(self):
processor_csv = DataProcessor(filepath="sample_dataset.csv")
processor_csv.save_summary(file_name="test_summary")
self.assertTrue(os.path.exists("test_summary.csv"))
def test_save_region_summary_plot(self):
processor = DataProcessor(filepath="sample_dataset.csv")
processor.save_region_summary_plot(file_name="test_plot")
self.assertTrue(os.path.exists("test_plot.png"))
def test_tear_down(self):
"""
Clean up temporary files created during the tests.
"""
try:
os.remove("test_summary.csv")
os.remove("test_summary.json")
os.remove("test_plot.png")
except FileNotFoundError:
pass
return super().tearDown()
if __name__ == "__main__":
unittest.main(verbosity=2)