geoffrey/eulaurarien
1
0
Fork 0
Code Issues 7 Pull requests Releases 1 Wiki Activity

Fix (most) mypy / flake8 errors

Browse source
This commit is contained in:
Geoffrey Frogeye 2021-08-14 23:35:51 +02:00
parent 3dcccad39a
commit f165e5a094
Signed by: geoffrey
GPG key ID: C72403E7F82E6AD8
4 changed files with 10 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
collect_subdomains.py
View file

@ -63,7 +63,7 @@ def collect_subdomains(url: str) -> typing.Iterable[str]:
for request in DRIVER.requests: for request in DRIVER.requests:
if request.response: if request.response:
yield subdomain_from_url(request.path) yield subdomain_from_url(request.path)
except: except Exception:
log.exception("Error") log.exception("Error")
DRIVER.quit() DRIVER.quit()
DRIVER = None DRIVER = None

5
feed_dns.py
View file

@ -37,7 +37,7 @@ class Writer(multiprocessing.Process):
if recs_queue: # MP if recs_queue: # MP
super(Writer, self).__init__() super(Writer, self).__init__()
self.recs_queue = recs_queue self.recs_queue = recs_queue
self.log = logging.getLogger(f"wr") self.log = logging.getLogger("wr")
self.autosave_interval = autosave_interval self.autosave_interval = autosave_interval
self.ip4_cache = ip4_cache self.ip4_cache = ip4_cache
if not recs_queue: # No MP if not recs_queue: # No MP
@ -45,7 +45,7 @@ class Writer(multiprocessing.Process):
def open_db(self) -> None: def open_db(self) -> None:
self.db = database.Database() self.db = database.Database()
self.db.log = logging.getLogger(f"wr") self.db.log = logging.getLogger("wr")
self.db.fill_ip4cache(max_size=self.ip4_cache) self.db.fill_ip4cache(max_size=self.ip4_cache)
def exec_record(self, record: Record) -> None: def exec_record(self, record: Record) -> None:
@ -74,6 +74,7 @@ class Writer(multiprocessing.Process):
block: typing.List[Record] block: typing.List[Record]
for block in iter(self.recs_queue.get, None): for block in iter(self.recs_queue.get, None):
assert block
record: Record record: Record
for record in block: for record in block:
self.exec_record(record) self.exec_record(record)

1
feed_rules.py
View file

@ -4,6 +4,7 @@ import database
import argparse import argparse
import sys import sys
import time import time
import typing
FUNCTION_MAP = { FUNCTION_MAP = {
"zone": database.Database.set_zone, "zone": database.Database.set_zone,

6
run_tests.py
View file

@ -57,7 +57,11 @@ if __name__ == "__main__":
perc_all = (100 * pass_all / count_all) if count_all else 100 perc_all = (100 * pass_all / count_all) if count_all else 100
perc_den = (100 * pass_den / count_den) if count_den else 100 perc_den = (100 * pass_den / count_den) if count_den else 100
log.info( log.info(
"%s: Entries %d/%d (%.2f%%) | Allow %d/%d (%.2f%%) | Deny %d/%d (%.2f%%)", (
"%s: Entries %d/%d (%.2f%%)"
" | Allow %d/%d (%.2f%%)"
"| Deny %d/%d (%.2f%%)"
),
filename, filename,
pass_ent, pass_ent,
count_ent, count_ent,