Commit ebd2504a authored by Romain Courteaud's avatar Romain Courteaud

Test empty status

parent 6fef2bb3
...@@ -152,8 +152,12 @@ class WebBot: ...@@ -152,8 +152,12 @@ class WebBot:
result_dict = OrderedDict() result_dict = OrderedDict()
# Report the bot status # Report the bot status
status = reportStatus(self._db).get()
result_dict["bot_status"] = [] result_dict["bot_status"] = []
try:
status = reportStatus(self._db).get()
except self._db.Status.DoesNotExist:
pass
else:
result_dict["bot_status"].append( result_dict["bot_status"].append(
{"text": status.text, "date": rfc822(status.timestamp)} {"text": status.text, "date": rfc822(status.timestamp)}
) )
...@@ -286,25 +290,7 @@ class WebBot: ...@@ -286,25 +290,7 @@ class WebBot:
} }
) )
if self.config["FORMAT"] == "json": return result_dict
print(json.dumps(result_dict))
else:
for table_key in result_dict:
print("# %s" % table_key)
print("")
table = result_dict[table_key]
if table:
# Print the header
table_key_list = [x for x in table[0].keys()]
table_key_list.sort()
print(" | ".join(table_key_list))
for line in table:
print(
" | ".join(
["%s" % (line[x]) for x in table_key_list]
)
)
print("")
def stop(self): def stop(self):
self._running = False self._running = False
...@@ -331,6 +317,7 @@ class WebBot: ...@@ -331,6 +317,7 @@ class WebBot:
raise raise
def run(self, mode): def run(self, mode):
status_dict = None
if mode not in ["crawl", "status"]: if mode not in ["crawl", "status"]:
raise NotImplementedError("Unexpected mode: %s" % mode) raise NotImplementedError("Unexpected mode: %s" % mode)
...@@ -344,13 +331,34 @@ class WebBot: ...@@ -344,13 +331,34 @@ class WebBot:
if mode in ["crawl", "all"]: if mode in ["crawl", "all"]:
self.crawl() self.crawl()
if mode in ["status", "all"]: if mode in ["status", "all"]:
self.status() status_dict = self.status()
except: except:
self.closeDB() self.closeDB()
raise raise
else: else:
self.closeDB() self.closeDB()
if status_dict is not None:
if self.config["FORMAT"] == "json":
print(json.dumps(status_dict))
else:
for table_key in status_dict:
print("# %s" % table_key)
print("")
table = status_dict[table_key]
if table:
# Print the header
table_key_list = [x for x in table[0].keys()]
table_key_list.sort()
print(" | ".join(table_key_list))
for line in table:
print(
" | ".join(
["%s" % (line[x]) for x in table_key_list]
)
)
print("")
def create_bot(**kw): def create_bot(**kw):
return WebBot(**kw) return WebBot(**kw)
...@@ -530,6 +530,37 @@ class SurykatkaBotTestCase(unittest.TestCase): ...@@ -530,6 +530,37 @@ class SurykatkaBotTestCase(unittest.TestCase):
checkHttpCodeChange(bot, []) checkHttpCodeChange(bot, [])
def test_status_emptyConfiguration(self):
resolver_ip = "192.168.0.254"
resolver = surykatka.dns.dns.resolver.Resolver(configure=False)
resolver.nameservers.append(resolver_ip)
with mock.patch(
"surykatka.configuration.get_default_resolver"
) as mock_get_default_resolver:
mock_get_default_resolver.return_value = resolver
bot = WebBot(mapping={"SQLITE": ":memory:"})
bot.initDB()
result = bot.status()
assert bot._db.Status.select().count() == 0
checkNetworkChange(bot, [])
checkDnsChange(bot, [])
checkSslChange(bot, [])
checkHttpCodeChange(bot, [])
assert result == {
"bot_status": [],
"dns_server": [],
"dns_query": [],
"http_server": [],
"ssl_certificate": [],
"http_query": [],
}
def suite(): def suite():
suite = unittest.TestSuite() suite = unittest.TestSuite()
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment