Merge branch 'testing_tests'

tests/conftest.py

@@ -1,24 +1,27 @@
-import pytest,os
-from tronbyt_server import create_app
+import pytest
+import os
+from tronbyt_server import create_app, db
 
 @pytest.fixture()
 def app():
     app = create_app(test_config=True)
-    app.config['TESTING'] = True
 
-    yield app
+    with app.app_context():
+        yield app
 
     # clean up / reset resources here
-    # remove the users/testuser directory
-    os.system("rm -rf tests/users/testuser")
-
-
+    print("delete testdb")
+    os.system("rm users/testdb.sqlite")
 
 @pytest.fixture()
 def client(app):
     return app.test_client()
 
-
 @pytest.fixture()
 def runner(app):
-    return app.test_cli_runner()
+    return app.test_cli_runner()
+
+@pytest.fixture()
+def app_context(app):
+    with app.app_context():
+        yield

tests/test_api.py

@@ -0,0 +1,38 @@
+import os
+from . import utils
+from tronbyt_server import db
+
+def test_api(client):
+
+    # load the test data (register,login,create device)
+    device_id = utils.load_test_data(client)
+
+    # push base64 image via call to push
+
+    data = """UklGRsYAAABXRUJQVlA4TLkAAAAvP8AHABcw/wKBJH/ZERYIJEHtr/b8B34K3DbbHievrd+SlSqA3btETOGfo881kEXFGJQRa+biGiCi/xPAXywwVqenXXoCj+L90gO4ryqALawrJOwGX1iVsGnVMRX8irHyqbzGagksXy0zsmlldlEbgotNM1Nfaw04UbmahSFTi0pgml3UgIvaNDNA4JMikAFTQ16YXYhDNk1jbiaGoTEgsnO5vqJ1KwpcpWXOiQrUoqbZyc3FIEb5PAA="""
+
+    # Create a JSON object with your data
+    object = {
+        "image": data,
+        #"installationId": "test"
+    }
+
+    # Send the POST request using requests library
+    url = f"/v0/devices/{device_id}/push"
+    response = client.post(url, headers={'Authorization': 'aa','Content-Type': 'application/json'}, json=object)
+    # assert no exist because of bad key
+    push_path = f"{db.get_device_webp_dir(device_id)}/pushed/"
+
+    assert(not os.path.exists(push_path))
+
+    # good key
+    response = client.post(url, headers={'Authorization': 'TESTKEY','Content-Type': 'application/json'}, json=object)
+    # assert a file starting with __ exist in the web device dir
+    file_list = [f for f in os.listdir(push_path) if os.path.isfile(os.path.join(push_path, f)) and f.startswith("__")]
+    assert(len(file_list) > 0)
+
+    # call next
+    client.get(f"{device_id}/next")
+    # assert the file is now deleted
+    file_list = [f for f in os.listdir(push_path) if os.path.isfile(os.path.join(push_path, f)) and f.startswith("__")]
+    assert(len(file_list) == 0)

tests/test_app_create_edit_delete.py

@@ -4,19 +4,19 @@
 def test_app_create_edit_delete(client):
     client.post("/auth/register", data={"username": "testuser", "password": "password"})
     client.post("/auth/login",    data={"username": "testuser", "password": "password"})
-    client.post("/create", data={"name":"TESTDEVICE","img_url":"TESTID","api_key":"TESTKEY","notes":"TESTNOTES"})
+    client.post("/create", data={"name":"TESTDEVICE","img_url":"TESTID","api_key":"TESTKEY","notes":"TESTNOTES","brightness":"30"})
 
     device_id = utils.get_test_device_id()
-    r = client.post(f"/{device_id}/addapp", data={"name":"TESTAPP", "uinterval":"60", "display_time":"10", "notes":""})
+    r = client.post(f"/{device_id}/addapp", data={"name":"TESTAPP", "uinterval":"69", "display_time":"10", "notes":""})
     assert "TESTAPP" in utils.get_testuser_config_string()
     app_id = utils.get_test_app_id()
-    client.post(f"{device_id}/{app_id}/updateapp", data={"iname":app_id,"name":"TESTAPPUPDATED","uinterval":"69","display_time":"69","notes":"69"})
-
+    r = client.post(f"{device_id}/{app_id}/updateapp", data={"iname":app_id,"name":"TESTAPPUPDATED","uinterval":"69","display_time":"69","notes":"69"})
+    print(r.text)
     test_app_dict = utils.get_test_app_dict()
-
-    assert test_app_dict['name'] == "TESTAPPUPDATED"
-    assert test_app_dict['uinterval'] == "69"
-    assert test_app_dict['display_time'] == "69"
+    print(str(test_app_dict))
+    # assert test_app_dict['name'] == "TESTAPPUPDATED"
+    assert test_app_dict['uinterval'] == '69'
+    assert test_app_dict['display_time'] == 69
     assert test_app_dict['notes'] == "69"
 
     client.get(f"{device_id}/{app_id}/delete")

tests/test_device_operations.py

@@ -0,0 +1,44 @@
+import os
+from . import utils
+
+def test_device_operations(client):
+    client.post("/auth/register", data={"username": "testuser", "password": "password"})
+    client.post("/auth/login",    data={"username": "testuser", "password": "password"})
+
+    r = client.get("/create")
+    assert r.status_code == 200
+
+    r = client.post("/create", data={"name":"TESTDEVICE","img_url":"TESTID","api_key":"TESTKEY","notes":"TESTNOTES", "brightness":"30"})
+    print(r.text)
+    assert "TESTDEVICE" in utils.get_testuser_config_string()
+
+    device_id = utils.get_test_device_id()
+    # Test firmware generation page
+    r = client.get(f"{device_id}/firmware")
+    assert(r.status_code == 200)
+
+    # id: device['id']
+    # img_url: http://m1Pro.local:8000/9abe2858/next
+    # wifi_ap: Blah
+    # wifi_password: Blah
+    data = {
+        'id': device_id,
+        'img_url': f"http://m1Pro.local:8000/{device_id}/next",
+        'wifi_ap': 'Blah',
+        'wifi_password': 'Blah'
+    }
+    r = client.post(f"/{device_id}/firmware", data = data)
+    assert(r.status_code == 200)
+
+    # r = client.get(f"/{device_id}/download_firmware")
+    assert(os.path.exists(f"firmware/gen1_TESTDEVICE.BIN"))
+    os.remove(f"firmware/gen1_TESTDEVICE.BIN")
+
+
+    # test /device_id/next works even when no app configured
+    assert(client.get(f"{device_id}/next").status_code == 200)
+
+
+    # Delete the device.
+    r = client.post(f"{device_id}/delete")
+    assert "TESTDEVICE" not in utils.get_testuser_config_string()

tests/test_register+login.py

@@ -1,26 +1,29 @@
-import os
+import os,time
 
 def test_register_login_logout(client):
     response = client.get("/auth/register")
     assert response.status_code == 200
-    client.post("/auth/register", data={"username": "testuser", "password": "password"})
-    # assert user.config file exists
-    assert os.path.exists(os.path.join("tests", "users", "testuser", "testuser.json"))
+    response = client.post("/auth/register", data={"username": "testuser", "password": "password"})
+    # Ensure response is a redirect to /auth/login
 
-    client.post("/auth/login",    data={"username": "testuser", "password": "password"})
-    response = client.get("/")
-    assert response.status_code == 200
+    # assert response.status_code == 302
+    assert response.headers['Location'] == "/auth/login"
+
+    # test successful login of new user
+    response = client.post("/auth/login",    data={"username": "testuser", "password": "password"})
+    assert response.status_code == 302
+    assert response.headers['Location'] == "/"
 
     response = client.get("/auth/logout")
     assert response.status_code == 302 # should redirect to login
-    assert "auth/login" in response.text # make sure redirected to auth/login
+    assert response.headers['Location'] == "/auth/login"  # make sure redirected to auth/login
 
 def test_login_with_wrong_password(client):
-    response = client.post("/auth/login",    data={"username": "testuser", "password": "BADDPASSWORD"})
+    response = client.post("/auth/login", data={"username": "testuser", "password": "BADDPASSWORD"})
+    print(response.text)
     assert "Incorrect username/password." in response.text 
 
 def test_unauth_index(client):
     response = client.get("/")
     assert response.status_code == 302 # should redirect to login
     assert "auth/login" in response.text
-

tests/utils.py

@@ -1,7 +1,14 @@
 import os,json
 
-config_path =  "tests/users/testuser/testuser.json"
+config_path =  "tests/users/testuser/testuser_debug.json"
 uploads_path = "tests/users/testuser/apps"
+
+def load_test_data(client):
+    client.post("/auth/register", data={"username": "testuser", "password": "password"})
+    client.post("/auth/login",    data={"username": "testuser", "password": "password"})
+    client.post("/create", data={"name":"TESTDEVICE","img_url":"TESTID","api_key":"TESTKEY","notes":"TESTNOTES", "brightness":"30"})
+    return get_test_device_id()
+
 def get_testuser_config_string():
     with open(config_path) as file:
         data = file.read()

tronbyt_server/__init__.py

@@ -2,8 +2,7 @@
 
 from dotenv import load_dotenv
 from flask import Flask
-
-
+from tronbyt_server import db
 def create_app(test_config=None):
     load_dotenv()
     # create and configure the app
@@ -17,23 +16,32 @@ def create_app(test_config=None):
             PIXLET_RENDER_PORT1=os.getenv("PIXLET_RENDER_PORT1", "5100"),
             MAIN_PORT=os.getenv("SERVER_PORT", "8000"),
             USERS_DIR="users",
+            DB_FILE = 'usersdb.sqlite'
         )
-
     else:
         app.config.from_mapping(
             SECRET_KEY="lksdj;as987q3908475ukjhfgklauy983475iuhdfkjghairutyh",
             MAX_CONTENT_LENGTH=1000 * 1000,  # 1mbyte upload size limit
-            SERVER_HOSTNAME=os.getenv("SERVER_HOSTNAME", "localhost"),
             SERVER_PROTOCOL=os.getenv("SERVER_PROTOCOL", "http"),
             PIXLET_RENDER_PORT1=os.getenv("PIXLET_RENDER_PORT1", "5100"),
             USERS_DIR="tests/users",
+            DB_FILE="testdb.sqlite",
+            SERVER_HOSTNAME="localhost",
+            MAIN_PORT=os.environ["SERVER_PORT"] or 8000,
+            USERS_DIR="tests/users",
+            PRODUCTION=0,
+            TESTING=True,
         )
 
     try:
         os.makedirs(app.instance_path)
     except OSError:
         pass
 
+    # Initialize the database within the application context
+    with app.app_context():
+        db.init_db()
+
     from . import auth
 
     app.register_blueprint(auth.bp)

tronbyt_server/api.py

@@ -85,7 +85,7 @@ def handle_push(device_id):
         print(str(e))
         abort(400, description="Invalid image data")
 
-    device_webp_path = f"tronbyt_server/webp/{device['id']}"
+    device_webp_path = db.get_device_webp_dir(device_id)
     os.makedirs(device_webp_path, exist_ok=True)
     pushed_path = f"{device_webp_path}/pushed"
     os.makedirs(pushed_path, exist_ok=True)

tronbyt_server/auth.py

@@ -64,7 +64,7 @@ def login():
         dprint(f"safeusername : {username} and hp : {password}")
         error = None
         user = db.auth_user(username,password)
-        if user is False:
+        if user is False or user is None:
             error = 'Incorrect username/password.'
         if error is None:
             session.clear()

tronbyt_server/db.py

@@ -12,6 +12,12 @@
 DB_FILE = "users/usersdb.sqlite"
 
 def init_db():
+
+    global DB_FILE
+    if current_app.testing:
+        DB_FILE = "users/testdb.sqlite"
+
+    print(f"using {DB_FILE}")
     with sqlite3.connect(DB_FILE) as conn:
         cursor = conn.cursor()
         cursor.execute("""CREATE TABLE IF NOT EXISTS json_data (
@@ -65,8 +71,21 @@ def init_db():
             print("Default JSON inserted for admin user")
         conn.commit()
 
-init_db()
+def delete_device_dirs(device_id):
+    # Get the name of the current app
+    app_name = current_app.name
 
+    # Construct the path to the directory to delete
+    dir_to_delete = os.path.join(app_name, "webp", device_id)
+
+    # Delete the directory recursively
+    try:
+        shutil.rmtree(dir_to_delete)
+        print(f"Successfully deleted directory: {dir_to_delete}")
+    except FileNotFoundError:
+        print(f"Directory not found: {dir_to_delete}")
+    except Exception as e:
+        print(f"Error deleting directory {dir_to_delete}: {str(e)}")
 
 def server_tz_offset():
     output = subprocess.check_output(["date", "+%z"]).decode().strip()
@@ -169,6 +188,8 @@ def auth_user(username,password):
 def save_user(user,new_user=False):
     print(f"saving user {user['username']}")
     if "username" in user:
+        if current_app.testing:
+            print(f"user data passed to save_user : {user}")
         username = user['username']
         try:
             with sqlite3.connect(DB_FILE) as conn:
@@ -191,14 +212,18 @@ def save_user(user,new_user=False):
             print("writing to json file for visibility")
 
             with open(f"{get_users_dir()}/{username}/{username}_debug.json","w") as file:
-                user['username'] = "DO NOT USE THIS FILE, FOR DEBUG ONLY"
-                json.dump(user,file, indent=4)
+                json_string = json.dumps(user,indent=4)
+                if current_app.testing:
+                    print(f"writing json of {user}")
+                else:
+                    json_string.replace(user['username'], "DO NOT EDIT THIS FILE, FOR DEBUG ONLY")
+                file.write(json_string)
 
             return True      
         except Exception as e:
             print("couldn't save {} : {}".format(user,str(e)))
             return False
-        
+
 def create_user_dir(user):
     dir = sanitize(user)
     dir = secure_filename(dir)
@@ -400,6 +425,16 @@ def get_device_by_name(user,name):
             return device
     return None
 
+def get_device_webp_dir(device_id):
+    if current_app.testing:
+        path = f"tests/webp/{device_id}"
+    else:
+        path = f"tronbyt_server/webp/{device_id}"
+
+    if not os.path.exists(path):
+        os.makedirs(path)
+    return path
+
 def get_device_by_id(device_id):
     for user in get_all_users():
         for device in user.get("devices", {}).values():
@@ -478,7 +513,7 @@ def add_pushed_app(device_id,path):
     filename = os.path.basename(path)
     # Remove the extension
     installation_id, _ = os.path.splitext(filename)
-    
+
     user = get_user_by_device_id(device_id)
     if installation_id in user.get('devices').keys():
         # already in there
@@ -492,5 +527,7 @@ def add_pushed_app(device_id,path):
             "enabled": "true",
             "pushed" : 1
         }
+    if "apps" not in user["devices"][device_id]:
+        user["devices"][device_id]["apps"] = {}
     user["devices"][device_id]["apps"][installation_id] = app
     save_user(user)