Refactored code into classes and added .gitignore

dev
Laura Nguyen 1 year ago
parent 4a8a7918b4
commit 63a0fae661
  1. 138
      .gitignore
  2. 28
      arch.py
  3. 3
      constants.py
  4. 31
      debian.py
  5. 25
      distro.py
  6. 30
      kernel.py
  7. 78
      main.py
  8. 28
      openbsd.py

138
.gitignore vendored

@ -0,0 +1,138 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/

@ -0,0 +1,28 @@
"""
Contains Arch class
"""
from datetime import datetime, timedelta
import requests
from distro import Distro
from constants import CSC_MIRROR
class Arch(Distro):
"""Arch class"""
@staticmethod
def name():
"""Get name of Arch"""
return "Arch"
@staticmethod
def check():
"""Check if Arch packages are up-to-date"""
arch_json = requests.get("https://archlinux.org/mirrors/status/json/").json()
last_check_str = arch_json["last_check"]
last_sync_str = [url for url in arch_json["urls"] if url["url"] == \
f"{CSC_MIRROR}archlinux/"][0]["last_sync"]
last_check_dt = datetime.strptime(last_check_str, "%Y-%m-%dT%H:%M:%S.%fZ")
last_sync_dt = datetime.strptime(last_sync_str, "%Y-%m-%dT%H:%M:%SZ")
# According to https://archlinux.org/mirrors/status/:
# Due to the timing of mirror checks, any value under one hour should be viewed as ideal
return last_check_dt < last_sync_dt + timedelta(hours = 1)

@ -0,0 +1,3 @@
"""Contains shared constants"""
CSC_MIRROR = "http://mirror.csclub.uwaterloo.ca/"

@ -0,0 +1,31 @@
"""
Contains Debian class
"""
from datetime import datetime, timedelta
import requests
from distro import Distro
from constants import CSC_MIRROR
class Debian(Distro):
"""Debian class"""
@staticmethod
def __get_dt(trace_file_url):
"""Get Debian datetime object from trace file"""
response_text = requests.get(trace_file_url).text
time_str = response_text.split('\n')[0]
return datetime.strptime(time_str, "%a %b %d %H:%M:%S UTC %Y")
@staticmethod
def name():
"""Get name of Debian"""
return "Debian"
@staticmethod
def check():
"""Check if Debian packages are up-to-date"""
official_dt = Debian.__get_dt("https://ftp-master.debian.org/debian/project/trace/master")
csc_dt = Debian.__get_dt(f"{CSC_MIRROR}debian/project/trace/master")
# Keep the table cell at https://mirror-master.debian.org/status/mirror-status.html
# green and not yellow
return official_dt < csc_dt + timedelta(hours = 7)

@ -0,0 +1,25 @@
"""
Contains abstract class for a distro
"""
from abc import ABC, abstractmethod
class Distro(ABC):
"""Abstract class for a distro"""
@staticmethod
@abstractmethod
def name():
"""Get name of distro"""
raise NotImplementedError
@staticmethod
@abstractmethod
def check():
"""Check if distro packages are up-to-date"""
raise NotImplementedError
@classmethod
def print_output(cls, is_successful):
"""Print final output of distro"""
output = "Success: {} up-to-date" if is_successful else "Failure: {} out-of-sync"
print(output.format(cls.name()))

@ -0,0 +1,30 @@
"""
Contains Kernel class
"""
import requests
from distro import Distro
from constants import CSC_MIRROR
class Kernel(Distro):
"""Kernel class"""
@staticmethod
def __get_line_count(checksum_file_url):
"""Get Kernel line count of checksum file"""
response_text = requests.get(checksum_file_url).text
return len(response_text.split('\n'))
@staticmethod
def name():
"""Get name of Kernel"""
return "Kernel"
@staticmethod
def check():
"""Check if Kernel packages are up-to-date"""
official_line_count = Kernel.__get_line_count(
"https://mirrors.edge.kernel.org/pub/linux/kernel/next/sha256sums.asc")
csc_line_count = Kernel.__get_line_count(
f"{CSC_MIRROR}kernel.org/linux/kernel/next/sha256sums.asc")
# A new update on a certain day adds 2 new lines
return official_line_count <= csc_line_count + 2

@ -4,75 +4,15 @@
This mirror status checker determines whether CSC mirror is up-to-date with upstream
"""
from datetime import datetime, timedelta
import requests
CSC_MIRROR = "http://mirror.csclub.uwaterloo.ca/"
SUCCESS = "Success: {} up-to-date"
FAILURE = "Failure: {} out-of-sync"
ERROR = "Error: {}\n{}"
def get_debian_dt(trace_file_url):
"""Get Debian datetime object from trace file"""
response_text = requests.get(trace_file_url).text
time_str = response_text.split('\n')[0]
return datetime.strptime(time_str, "%a %b %d %H:%M:%S UTC %Y")
def get_kernel_line_count(checksum_file_url):
"""Get Kernel line count of checksum file"""
response_text = requests.get(checksum_file_url).text
return len(response_text.split('\n'))
def get_openbsd_sec(timestamp_file_url):
"""Get OpenBSD seconds since the Epoch from timestamp file"""
sec_str = requests.get(timestamp_file_url).text
return int(sec_str)
from arch import Arch
from debian import Debian
from kernel import Kernel
from openbsd import OpenBSD
if __name__ == "__main__":
# Arch
try:
arch_json = requests.get("https://archlinux.org/mirrors/status/json/").json()
last_check_str = arch_json["last_check"]
last_sync_str = [url for url in arch_json["urls"] if url["url"] == \
f"{CSC_MIRROR}archlinux/"][0]["last_sync"]
last_check_dt = datetime.strptime(last_check_str, "%Y-%m-%dT%H:%M:%S.%fZ")
last_sync_dt = datetime.strptime(last_sync_str, "%Y-%m-%dT%H:%M:%SZ")
# According to https://archlinux.org/mirrors/status/:
# Due to the timing of mirror checks, any value under one hour should be viewed as ideal
OUTPUT = SUCCESS if last_check_dt < last_sync_dt + timedelta(hours = 1) else FAILURE
print(OUTPUT.format("Arch"))
except requests.exceptions.RequestException as err:
print(ERROR.format("Arch", err))
# Debian
try:
official_dt = get_debian_dt("https://ftp-master.debian.org/debian/project/trace/master")
csc_dt = get_debian_dt(f"{CSC_MIRROR}debian/project/trace/master")
# Keep the table cell at https://mirror-master.debian.org/status/mirror-status.html
# green and not yellow
OUTPUT = SUCCESS if official_dt < csc_dt + timedelta(hours = 7) else FAILURE
print(OUTPUT.format("Debian"))
except requests.exceptions.RequestException as err:
print(ERROR.format("Debian", err))
# Kernel
try:
official_line_count = get_kernel_line_count(
"https://mirrors.edge.kernel.org/pub/linux/kernel/next/sha256sums.asc")
csc_line_count = get_kernel_line_count(
f"{CSC_MIRROR}kernel.org/linux/kernel/next/sha256sums.asc")
# A new update on a certain day adds 2 new lines
OUTPUT = SUCCESS if official_line_count <= csc_line_count + 2 else FAILURE
print(OUTPUT.format("Kernel"))
except requests.exceptions.RequestException as err:
print(ERROR.format("Kernel", err))
# OpenBSD
try:
official_sec = get_openbsd_sec("https://ftp.openbsd.org/pub/OpenBSD/timestamp")
csc_sec = get_openbsd_sec(f"{CSC_MIRROR}OpenBSD/timestamp")
# Out-of-sync by 1 day maximum
OUTPUT = SUCCESS if official_sec < csc_sec + 86400 else FAILURE
print(OUTPUT.format("OpenBSD"))
except requests.exceptions.RequestException as err:
print(ERROR.format("OpenBSD", err))
for distro in [Arch, Debian, Kernel, OpenBSD]:
try:
distro.print_output(distro.check())
except requests.exceptions.RequestException as err:
print(f"Error: {distro.name()}\n{err}")

@ -0,0 +1,28 @@
"""
Contains OpenBSD class
"""
import requests
from distro import Distro
from constants import CSC_MIRROR
class OpenBSD(Distro):
"""OpenBSD class"""
@staticmethod
def __get_sec(timestamp_file_url):
"""Get OpenBSD seconds since the Epoch from timestamp file"""
sec_str = requests.get(timestamp_file_url).text
return int(sec_str)
@staticmethod
def name():
"""Get name of OpenBSD"""
return "OpenBSD"
@staticmethod
def check():
"""Check if OpenBSD packages are up-to-date"""
official_sec = OpenBSD.__get_sec("https://ftp.openbsd.org/pub/OpenBSD/timestamp")
csc_sec = OpenBSD.__get_sec(f"{CSC_MIRROR}OpenBSD/timestamp")
# Out-of-sync by 1 day maximum
return official_sec < csc_sec + 86400
Loading…
Cancel
Save