import unittest import mock import context from steam_review_scraper import db_common, review_model import time import datetime db_common.testing_mode = True def insert_dummy_data(conn, appid): columns = [ "id", "recommended", "review_url", "date_posted", "steam_appid", "lang_key", "entry_updated" ] update_time = datetime.datetime.fromtimestamp(10) posted_time = datetime.datetime.fromtimestamp(10) item1 = (1, True, "", posted_time, appid, "english", update_time) item2 = (2, False, "", posted_time, appid, "french", update_time) item3 = (3, True, "", posted_time, 200000, "english", update_time) data = [item1, item2, item3] conn.executemany( "INSERT INTO {0} ({1}) VALUES ({2})".format( db_common.db_definition.TABLE_STEAM_REVIEWS, ", ".join(columns), ", ".join(["?"] * len(columns)), ), data) def get_dummy_steam_review(appid, id): return review_model.SteamReview( id, "", appid, recommended = True, date_posted = datetime.datetime.fromtimestamp(10), lang_key = "english", can_be_turned = True ) def insert_steam_review(conn, review): columns = db_common.db_definition.COLUMNS_STEAM_REVIEWS.keys() data = ( review.id, review.steam_appid, review.recommended, review.user_name, review.content, review.hours_played, review.review_url, review.date_posted, review.date_updated, review.helpful_amount, review.helpful_total, review.games_owned, review.responded_by, review.responded_date, review.lang_key, review.received_compensation, time.time(), review.can_be_turned, review.issue_list ) conn.execute( "INSERT INTO {0} ({1}) VALUES ({2})".format( db_common.db_definition.TABLE_STEAM_REVIEWS, ", ".join(columns), ", ".join(["?"] * len(columns)), ), data) def get_last_steam_review(conn): data = conn.execute( "SELECT {0} FROM {1} ORDER BY entry_updated DESC".format( ", ".join(db_common.db_definition.COLUMNS_STEAM_REVIEWS.keys()), db_common.db_definition.TABLE_STEAM_REVIEWS) ).fetchall()[0] return db_common.steam_review_from_result(data) class GetTotalReviewCountTest(unittest.TestCase): def setUp(self): self.appid = 000000 self.test_conn = db_common.create_database(":memory:") insert_dummy_data(self.test_conn, self.appid) @mock.patch("sqlite3.connect") def test_filter_appid(self, conn_mock): conn_mock.return_value = self.test_conn count = db_common.get_review_count(self.appid) self.assertEqual(count, 2) @mock.patch("sqlite3.connect") def test_filter_languages(self, conn_mock): conn_mock.return_value = self.test_conn review_query = db_common.ReviewQuery(lang_key="english") count = db_common.get_review_count(self.appid, review_query) self.assertEqual(count, 1) @mock.patch("sqlite3.connect") def test_filter_recommended(self, conn_mock): conn_mock.return_value = self.test_conn review_query = db_common.ReviewQuery(recommended=False) count = db_common.get_review_count(self.appid, review_query) self.assertEqual(count, 1) class GetReviewsTest(unittest.TestCase): def setUp(self): self.appid = 000000 self.test_conn = db_common.create_database(":memory:") insert_dummy_data(self.test_conn, self.appid) @mock.patch("sqlite3.connect") def test_filter_appid(self, conn_mock): conn_mock.return_value = self.test_conn reviews = db_common.get_reviews(self.appid) self.assertEqual(len(reviews), 2) @mock.patch("sqlite3.connect") def test_filter_languages(self, conn_mock): conn_mock.return_value = self.test_conn review_query_settings = db_common.ReviewQuery( lang_key="english") reviews = db_common.get_reviews( self.appid, review_query_object=review_query_settings ) self.assertEqual(len(reviews), 1) @mock.patch("sqlite3.connect") def test_advanced_filtering(self, conn_mock): conn_mock.return_value = self.test_conn item = get_dummy_steam_review(self.appid, 100) item.date_posted = datetime.datetime.fromtimestamp(4) item.date_updated = datetime.datetime.fromtimestamp(10) item.responded_by = 5000 item.responded_date = datetime.datetime.fromtimestamp(5) item.can_be_turned = True item.recommended = True insert_steam_review(self.test_conn, item) review_query_settings = db_common.ReviewQuery( lang_key="english", from_date=datetime.datetime.fromtimestamp(2), until_date=datetime.datetime.fromtimestamp(12), hide_never_updated=True, only_update_after_response=True, can_be_turned=True, has_response=True, responded_by=5000, recommended=True, ) reviews = db_common.get_reviews( self.appid, review_query_object=review_query_settings ) self.assertEqual(len(reviews), 1) @mock.patch("sqlite3.connect") def test_only_resolved_issues(self, conn_mock): conn_mock.return_value = self.test_conn item1 = get_dummy_steam_review(self.appid, 100) item1.issue_list = [1000, 2000] item2 = get_dummy_steam_review(self.appid, 101) item2.issue_list = [2000, 3000, 4000] insert_steam_review(self.test_conn, item1) insert_steam_review(self.test_conn, item2) columns = db_common.db_definition.COLUMNS_STEAM_REVIEW_ISSUES.keys() item1 = (1000, "issue1", 10) item2 = (2000, "issue2", 1) item3 = (3000, "issue2", 0) # Unresolved. data = [item1, item2, item3] self.test_conn.executemany( "INSERT INTO {0} ({1}) VALUES ({2})".format( db_common.db_definition.TABLE_STEAM_REVIEW_ISSUES, ", ".join(columns), ", ".join(["?"] * len(columns)), ), data) review_query_settings = db_common.ReviewQuery( only_resolved_issues=True ) reviews = db_common.get_reviews( self.appid, review_query_object=review_query_settings ) self.assertEqual(len(reviews), 1) self.assertEqual(reviews[0].id, 100) @mock.patch("sqlite3.connect") def test_filter_issue_list(self, conn_mock): conn_mock.return_value = self.test_conn issue_list = [ 1000, 2000, 3000 ] item1 = get_dummy_steam_review(self.appid, 100) item1.issue_list = issue_list item2 = get_dummy_steam_review(self.appid, 101) item2.issue_list = [1000] item3 = get_dummy_steam_review(self.appid, 102) item3.issue_list = [5000] insert_steam_review(self.test_conn, item1) insert_steam_review(self.test_conn, item2) review_query_settings = db_common.ReviewQuery( only_resolved_issues=False, issue_list=issue_list ) reviews = db_common.get_reviews( self.appid, review_query_object=review_query_settings ) self.assertEqual(len(reviews), 2) self.assertListEqual(reviews[0].issue_list, issue_list) @mock.patch("sqlite3.connect") def test_page_number(self, conn_mock): conn_mock.return_value = self.test_conn reviews = db_common.get_reviews( self.appid, page_number=1, reviews_per_page=1, ) self.assertEqual(len(reviews), 1) self.assertEqual(reviews[0].id, 2) @mock.patch("sqlite3.connect") def test_sorted_id(self, conn_mock): conn_mock.return_value = self.test_conn reviews = db_common.get_reviews( self.appid, sort_by="id", sort_order=db_common.db_definition.ORDER_MODES[0] ) self.assertEqual(reviews[0].id, 2) self.assertEqual(reviews[-1].id, 1) class InsertOrUpdateReviewsTest(unittest.TestCase): def setUp(self): self.appid = 000000 self.test_conn = db_common.create_database(":memory:") def get_count(self): return self.test_conn.execute( "SELECT count(id) FROM {}".format( db_common.db_definition.TABLE_STEAM_REVIEWS) ).fetchall()[0][0] def get_last(self): return self.test_conn.execute( "SELECT id,steam_appid FROM {} ORDER BY entry_updated DESC".format( db_common.db_definition.TABLE_STEAM_REVIEWS) ).fetchall()[0] @mock.patch("test_db_common.db_common.get_db_connection") def test_insert_single(self, conn_mock): conn_mock.return_value = self.test_conn item = get_dummy_steam_review(self.appid, 100) update_time = int(time.time()) db_common.insert_or_update_reviews([item], update_time) count = self.get_count() self.assertEqual(count, 1) @mock.patch("test_db_common.db_common.get_db_connection") def test_insert_multiple(self, conn_mock): conn_mock.return_value = self.test_conn items = [get_dummy_steam_review(self.appid, 100+id) for id in range(5)] update_time = int(time.time()) db_common.insert_or_update_reviews(items, update_time) count = self.get_count() self.assertEqual(count, 5) @mock.patch("test_db_common.db_common.get_db_connection") def test_include_user_input_columns(self, conn_mock): conn_mock.return_value = self.test_conn issue_list = [1, 2] item = get_dummy_steam_review(self.appid, 100) item.can_be_turned = True item.issue_list = issue_list update_time = datetime.datetime.fromtimestamp(10) db_common.insert_or_update_reviews( [item], update_time, include_user_input_columns=True) steam_review = get_last_steam_review(self.test_conn) self.assertEqual(steam_review.can_be_turned, True) self.assertListEqual(steam_review.issue_list, issue_list) @mock.patch("test_db_common.db_common.get_db_connection") def test_content(self, conn_mock): conn_mock.return_value = self.test_conn id = 100 item = get_dummy_steam_review(self.appid, id) update_time = int(time.time()) db_common.insert_or_update_reviews([item], update_time) review = self.get_last() self.assertEqual(review[0], id) self.assertEqual(review[1], self.appid) class DeleteAllUnchangedReviewsTest(unittest.TestCase): def setUp(self): self.appid = 000000 self.test_conn = db_common.create_database(":memory:") def get_dummy_steam_review(self, id): return review_model.SteamReview( id, "", self.appid, recommended = True, date_posted = 10, lang_key = "english" ) @mock.patch("test_db_common.db_common.get_db_connection") def test_single_deleted(self, conn_mock): conn_mock.return_value = self.test_conn compare_time = 10 update_time = 8 languages = ["english"] insert_dummy_data(self.test_conn, self.appid) item = self.get_dummy_steam_review(100) db_common.insert_or_update_reviews([item], update_time) deleted_count = db_common.delete_all_unchanged_reviews( self.appid, languages, compare_time) self.assertEqual(deleted_count, 1) @mock.patch("test_db_common.db_common.get_db_connection") def test_none_deleted(self, conn_mock): conn_mock.return_value = self.test_conn compare_time = 10 languages = ["english"] insert_dummy_data(self.test_conn, self.appid) deleted_count = db_common.delete_all_unchanged_reviews( self.appid, languages, compare_time) self.assertEqual(deleted_count, 0) if __name__ == "__main__": unittest.main()