89 lines
2.3 KiB
Python
89 lines
2.3 KiB
Python
import logging
|
|
import subprocess
|
|
import sys
|
|
import os
|
|
import auth.api_security as auth
|
|
import time
|
|
import threading
|
|
from tests import db_test
|
|
|
|
try:
|
|
os.remove("api.log")
|
|
os.remove("api_server.log")
|
|
os.remove("app.log")
|
|
except Exception as e:
|
|
logging.error(e)
|
|
#Logging
|
|
logging.basicConfig(
|
|
filename='app.log', # Log to a file
|
|
level=logging.DEBUG, # Log everything (DEBUG and above)
|
|
format='%(asctime)s %(levelname)s %(message)s'
|
|
)
|
|
|
|
logging.info('Application started')
|
|
try:
|
|
logging.debug('Debug info here')
|
|
except Exception as e:
|
|
logging.exception('An error occurred')
|
|
|
|
|
|
env = os.environ.copy()
|
|
env["PYTHONIOENCODING"] = "utf-8"
|
|
|
|
try:
|
|
os.remove("migration_status.flag")
|
|
except Exception as e:
|
|
logging.info(e)
|
|
#migration
|
|
def run_migration():
|
|
global migration_done
|
|
process = subprocess.Popen(
|
|
["python", "migration/alembic_scipt.py"],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL
|
|
)
|
|
process.wait()
|
|
migration_done = (process.returncode == 0)
|
|
with open("flags/migration_status.flag", "w") as f:
|
|
f.write("migration_done" if migration_done else "failed")
|
|
|
|
|
|
|
|
#DB test
|
|
def run_db_test():
|
|
tokens = db_test.test_table("tokens", ["tokens", "time", "end_date"])
|
|
user = db_test.test_table("user", ["username", "password", "role"])
|
|
items = db_test.test_table("luggage", ["clothes", "weight", "pieces"])
|
|
time.sleep(5)
|
|
if tokens == True & user == True & items == True:
|
|
with open("flags/db.flag", "w") as f:
|
|
f.write("db_check_done")
|
|
else:
|
|
os._exit(1)
|
|
|
|
#cli and api
|
|
log_file = open("api_server.log", "w", encoding="utf-8")
|
|
api_process = subprocess.Popen(
|
|
["fastapi", "dev", "api_server.py", "--reload-dir", "web", "--reload-dir", "api", "--reload-dir", "api_server.py",],
|
|
stdout=log_file,
|
|
stderr=log_file,
|
|
env=env
|
|
)
|
|
cli_process = subprocess.Popen([sys.executable, "cli.py"])
|
|
|
|
db_thread = threading.Thread(target=run_db_test)
|
|
db_thread.start()
|
|
thread = threading.Thread(target=run_migration)
|
|
thread.start()
|
|
threading.Thread(target=api_process.wait, daemon=True).start()
|
|
threading.Thread(target=cli_process.wait, daemon=True).start()
|
|
|
|
# automatic durataion validation check
|
|
while True:
|
|
time.sleep(300)
|
|
try:
|
|
auth.check_validaty()
|
|
|
|
except Exception as e:
|
|
logging.info(e)
|