Newspaper-Topic-Modelling/data/src/tests/test_data_processor.py at main · CZboop/Newspaper-Topic-Modelling · GitHub

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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
import unittest, pytest
from pathlib import Path
import sys
sys.path.append(f"{Path(__file__).parent.parent}")
from data_processor import DataProcessor
import os
from pathlib import Path
import pandas as pd
import pandas.testing as pd_testing
import datetime
import shutil

# @unittest.skip('skipping for speed while still writing other new tests')
class TestDataProcessor(unittest.TestCase):

    maxDiff = None

    # setup before the test suite run
    @classmethod
    def setUpClass(self):
        # create temp directory for files with test data
        self.test_dir_name = 'temp_test_files'
        self.temp_within_current_dir = f'{Path(__file__).parent}/{self.test_dir_name}'
        Path(self.temp_within_current_dir).mkdir(parents=True, exist_ok=True)

    def setup_write_test_csv_file(self, dataframe, name):
        dataframe.to_csv(f'{self.temp_within_current_dir}/{name}')

    def test_select_data_files(self):
        # given - a data processor object with selector for csv files starting with test, path pointing to the temp test files directory
        for i in range(5):
            test_dataframe = pd.DataFrame(data= {'headline' : ['testing', 'test of the test', 'another test', 'final test'], 'url' : ['www.test-news.co.uk/123', 'www.test-news.co.uk/321', 'www.test-news.co.uk/3141', 'www.test-news.co.uk/897']})
            self.setup_write_test_csv_file(test_dataframe, f'test_{i}.csv')

        data_processor = DataProcessor(f'./{self.test_dir_name}', ['headline', 'url'], selector= 'test*.csv')

        # when - we call the select data files method of the data processor
        actual = map(lambda x: os.path.normpath(x), data_processor.select_data_files())

        # then - an array of the five test files is returned, and the data processor has a new property called files
        expected = map(lambda x: os.path.normpath(x), [f'./{self.test_dir_name}/test_0.csv', f'./{self.test_dir_name}/test_1.csv', f'./{self.test_dir_name}/test_2.csv', f'./{self.test_dir_name}/test_3.csv', f'./{self.test_dir_name}/test_4.csv'])

        self.assertEqual(sorted(actual), sorted(expected))
        self.assertTrue(hasattr(data_processor, 'files'))

    def test_read_and_concat_data_files(self):
        # given - a data processor object with parameters to read in dummy files with names starting with test
        all_dfs = []
        for i in range(5):
            test_dataframe = pd.DataFrame(data= {'headline' : ['testing', 'test of the test', 'another test', 'final test'], 'url' : ['www.test-news.co.uk/123', 'www.test-news.co.uk/321', 'www.test-news.co.uk/3141', 'www.test-news.co.uk/897'], 'source' : ['test', 'test', 'test', 'test']})
            all_dfs.append(test_dataframe)
            self.setup_write_test_csv_file(test_dataframe, f'test_{i}.csv')

        data_processor = DataProcessor(f'./{self.test_dir_name}', ['headline', 'url'], selector= 'test*.csv')

        # when - we call the read and concat data files method
        actual = data_processor.read_and_concat_data_files()

        # then - a pandas dataframe of combined data from all the files is returned, with a new column 'source' based on the data selector/file names
        expected_df = pd.concat(all_dfs, axis=0, ignore_index=True)

        pd_testing.assert_frame_equal(actual, expected_df)
        #  - and the pandas dataframe is assigned to a class attribute 'combined_data'
        self.assertTrue(hasattr(data_processor, 'combined_data'))

    def test_clean_text_lowercases_text(self):
        # given - a data processor class and mixed case data
        data_processor = DataProcessor(f'./{self.test_dir_name}', ['headline', 'url'], selector= 'test*.csv')
        headline = 'Testing Test of The Test Another Test Final Test'

        # when - we call clean_text on the headline
        actual = data_processor._clean_text(headline)

        # then - the resulting string is the same if it is lowercased
        self.assertEqual(actual, actual.lower())

    def test_clean_text_removes_stopwords(self):
        # given - a data processor class and lowercase data with some stopwords
        data_processor = DataProcessor(f'./{self.test_dir_name}', ['headline', 'url'], selector= 'test*.csv')
        headline = 'testing test of the test and a test final test'

        # when - we call clean_text on the headline
        actual = data_processor._clean_text(headline)

        # then - the resulting string has stopwords removed
        expected = 'testing test test test final test'
        self.assertEqual(actual, expected)

    def test_remove_duplicates_and_nones_can_remove_duplicates(self):
        # given - a data processor set up with duplicate data in the headlines
        for i in range(5):
            test_dataframe = pd.DataFrame(data= {'headline' : ['testing', 'test of the test', 'another test', 'final test'], 'url' : ['www.test-news.co.uk/123', 'www.test-news.co.uk/321', 'www.test-news.co.uk/3141', 'www.test-news.co.uk/897'], 'source' : ['test', 'test', 'test', 'test']})
            self.setup_write_test_csv_file(test_dataframe, f'test_{i}.csv')

        data_processor = DataProcessor(f'./{self.test_dir_name}', ['headline', 'url'], selector= 'test*.csv')

        # when - we call remove duplicates and nones, and extract the column that should have duplicates dropped
        actual = list(data_processor.remove_duplicates_and_nones()['headline'])
        expected = ['testing', 'test of the test', 'another test', 'final test']

        # then - the returned data does not have duplicates
        self.assertEqual(sorted(actual), sorted(expected))

    def test_remove_duplicates_and_nones_can_remove_nones(self):
        # given - a data processor set up with some missing data of different types (but no duplicates)
        headline_list_of_lists = [['testingtesting', '', 'another test', None], [None, 'test of the test', '', 'final test'], ['testing test', 'test of the other test', None, 'final test but different'], ['testing 123', None, 'a different test', ''], ['testing', None, None, 'final test the final one']]
        for i in range(5):
            test_dataframe = pd.DataFrame(data= {'headline' : headline_list_of_lists[i], 'url' : ['www.test-news.co.uk/123', 'www.test-news.co.uk/321', 'www.test-news.co.uk/3141', 'www.test-news.co.uk/897'], 'source' : ['test', 'test', 'test', 'test']})
            self.setup_write_test_csv_file(test_dataframe, f'test_{i}.csv')

        data_processor = DataProcessor(f'./{self.test_dir_name}', ['headline', 'url'], selector= 'test*.csv')

        # when - we call remove duplicates and nones, and extract the column that should have duplicates removed
        actual = list(data_processor.remove_duplicates_and_nones()['headline'])
        expected = ['testingtesting', 'another test', 'test of the test', 'final test', 'testing test', 'test of the other test', 'final test but different', 'testing 123', 'a different test', 'testing','final test the final one']

        # then - the returned data is the original minus the missing data
        self.assertEqual(sorted(actual), sorted(expected))

    def test_filter_dates_removes_dates_before_start_date(self):
        # given - a data processor with start/end date parameters passed in, and data containing dates with some dates outside the start/end range
        headline_list_of_lists = [['testingtesting', 'a test of the test', 'another test', 'test headline'], ['one of the tests', 'test of the test', 'a headline', 'final test'], ['testing test', 'test of the other test', 'a headline with a value', 'final test but different'], ['testing 123', 'here is a test', 'a different test', 'there is another test'], ['testing', 'it is a test', 'it is not a test', 'final test the final one']]

        date_list_of_lists = [[ datetime.date(2019, 12, 1), datetime.date(2019, 11, 1), datetime.date(2029, 12, 1), datetime.date(2019, 12, 21)], [ datetime.date(2022, 12, 1), datetime.date(2022, 10, 1), datetime.date(2020, 5, 12), datetime.date(2021, 3, 9)], [ datetime.date(2032, 12, 1), datetime.date(2025, 8, 14), datetime.date(2122, 9, 14), datetime.date(2022, 2, 11)], [ datetime.date(2021, 7, 12), datetime.date(2021, 8, 14), datetime.date(1999, 12, 1), datetime.date(2009, 7, 12)], [ datetime.date(2020, 9, 8), datetime.date(2020, 5, 29), datetime.date(2029, 7, 23), datetime.date(2020, 1, 18)]]

        for i in range(5):
            test_dataframe = pd.DataFrame(data= {'headline' : headline_list_of_lists[i], 'url' : ['www.test-news.co.uk/123', 'www.test-news.co.uk/321', 'www.test-news.co.uk/3141', 'www.test-news.co.uk/897'], 'date': date_list_of_lists[i]})
            self.setup_write_test_csv_file(test_dataframe, f'test_{i}.csv')

        start_date = datetime.date(2019, 12, 1)
        end_date = datetime.date(2022, 12, 1)

        data_processor = DataProcessor(f'./{self.test_dir_name}', ['headline', 'url', 'date'], selector= 'test*.csv', start_date= start_date, end_date= end_date)

        # when - the filter dates method is called and the date column is extracted
        actual = list(data_processor.filter_dates()['date'])

        # then - there are no dates before the start date
        actual_before_start_date = list(filter(lambda x: x < start_date, actual))

        # and there are no dates after the end date
        self.assertEqual(len(actual_before_start_date), 0)

    def test_filter_dates_removes_dates_after_end_date(self):
        # given - a data processor with start/end date parameters passed in, and data containing dates with some dates outside the start/end range
        headline_list_of_lists = [['testingtesting', 'a test of the test', 'another test', 'test headline'], ['one of the tests', 'test of the test', 'a headline', 'final test'], ['testing test', 'test of the other test', 'a headline with a value', 'final test but different'], ['testing 123', 'here is a test', 'a different test', 'there is another test'], ['testing', 'it is a test', 'it is not a test', 'final test the final one']]

        date_list_of_lists = [[ datetime.date(2019, 12, 1), datetime.date(2019, 11, 1), datetime.date(2029, 12, 1), datetime.date(2019, 12, 21)], [ datetime.date(2022, 12, 1), datetime.date(2022, 10, 1), datetime.date(2020, 5, 12), datetime.date(2021, 3, 9)], [ datetime.date(2032, 12, 1), datetime.date(2025, 8, 14), datetime.date(2122, 9, 14), datetime.date(2022, 2, 11)], [ datetime.date(2021, 7, 12), datetime.date(2021, 8, 14), datetime.date(1999, 12, 1), datetime.date(2009, 7, 12)], [ datetime.date(2020, 9, 8), datetime.date(2020, 5, 29), datetime.date(2029, 7, 23), datetime.date(2020, 1, 18)]]

        for i in range(5):
            test_dataframe = pd.DataFrame(data= {'headline' : headline_list_of_lists[i], 'url' : ['www.test-news.co.uk/123', 'www.test-news.co.uk/321', 'www.test-news.co.uk/3141', 'www.test-news.co.uk/897'], 'date': date_list_of_lists[i]})
            self.setup_write_test_csv_file(test_dataframe, f'test_{i}.csv')

        start_date = datetime.date(2019, 12, 1)
        end_date = datetime.date(2022, 12, 1)

        data_processor = DataProcessor(f'./{self.test_dir_name}', ['headline', 'url', 'date'], selector= 'test*.csv', start_date= start_date, end_date= end_date)

        # when - the filter dates method is called and the date column is extracted
        actual = list(data_processor.filter_dates()['date'])

        # then - there are no dates before the start date
        actual_after_end_date = list(filter(lambda x: x > end_date, actual))

        # and there are no dates after the end date
        self.assertEqual(len(actual_after_end_date), 0)

    def test_filter_topics(self):
        # given - some data with different topics in url and a data processor object set to filter out one of the topics
        headline_list_of_lists = [['testingtesting', 'a test of the test', 'another test', 'test headline'], ['one of the tests', 'test of the test', 'a headline', 'final test'], ['testing test', 'test of the other test', 'a headline with a value', 'final test but different'], ['testing 123', 'here is a test', 'a different test', 'there is another test'], ['testing', 'it is a test', 'it is not a test', 'final test the final one']]

        url_list_of_lists = [[ 'www.test-news/science/article001', 'www.test-news/politics/article002', 'www.test-news/tech/article003', 'www.test-news/education/article004'], [ 'www.test-news/politics/article005', 'www.test-news/tech/article006', 'www.test-news/politics/article007', 'www.test-news/science/article008'], [ 'www.test-news/science/article009', 'www.test-news/politics/article010', 'www.test-news/tech/article011', 'www.test-news/politics/article012'], [ 'www.test-news/tech/article013', 'www.test-news/politics/article014', 'www.test-news/tech/article015', 'www.test-news/politics/article016'], [ 'www.test-news/tech/article017', 'www.test-news/politics/article018', 'www.test-news/science/article019', 'www.test-news/politics/article020']]

        for i in range(5):
            test_dataframe = pd.DataFrame(data= {'headline' : headline_list_of_lists[i], 'url' : url_list_of_lists[i], 'date': [ datetime.date(2019, 12, 1), datetime.date(2019, 11, 1), datetime.date(2029, 12, 1), datetime.date(2019, 12, 21)]})
            self.setup_write_test_csv_file(test_dataframe, f'test_{i}.csv')

        data_processor = DataProcessor(f'./{self.test_dir_name}', ['headline', 'url', 'date'], selector= 'test*.csv', topics_to_remove= ['science'])

        # when - we call filter topics and extract the url column
        actual = list(data_processor.filter_topics()['url'])

        # then - resulting data no longer includes the topic that should be filtered out
        expected = ['www.test-news/politics/article002', 'www.test-news/tech/article003', 'www.test-news/education/article004','www.test-news/politics/article005', 'www.test-news/tech/article006', 'www.test-news/politics/article007', 'www.test-news/politics/article010', 'www.test-news/tech/article011', 'www.test-news/politics/article012','www.test-news/tech/article013', 'www.test-news/politics/article014', 'www.test-news/tech/article015', 'www.test-news/politics/article016','www.test-news/tech/article017', 'www.test-news/politics/article018', 'www.test-news/politics/article020']
        self.assertEqual(sorted(actual), sorted(expected))

    # teardown to undo temp changes after the test suite run - removing temporary test files and directory
    @classmethod
    def tearDownClass(self):
        try:
            shutil.rmtree(self.temp_within_current_dir)
        except OSError as error:
            print(f'An error occured while trying to delete directory: {error.filename} - {error.strerror}.')

if __name__ == "__main__":
    unittest.main()