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()