Usblog l2 60497 backup

.github/workflows/L2-tests.yml

@@ -31,11 +31,11 @@ jobs:
 
       - name: Start mock-xconf service
         run: |
-          docker run -d --name mockxconf -p 50050:50050 -p 50051:50051 -p 50052:50052 -v ${{ github.workspace }}:/mnt/L2_CONTAINER_SHARED_VOLUME ghcr.io/rdkcentral/docker-device-mgt-service-test/mockxconf:latest
+          docker run -d --name mockxconf -p 50050:50050 -p 50051:50051 -p 50052:50052 -e ENABLE_MTLS=true -v ${{ github.workspace }}:/mnt/L2_CONTAINER_SHARED_VOLUME ghcr.io/rdkcentral/docker-device-mgt-service-test/mockxconf:latest
 
       - name: Start l2-container service
         run: |
-          docker run -d --name native-platform --link mockxconf -v ${{ github.workspace }}:/mnt/L2_CONTAINER_SHARED_VOLUME ghcr.io/rdkcentral/docker-device-mgt-service-test/native-platform:latest
+          docker run -d --name native-platform --link mockxconf -e ENABLE_MTLS=true -v ${{ github.workspace }}:/mnt/L2_CONTAINER_SHARED_VOLUME ghcr.io/rdkcentral/docker-device-mgt-service-test/native-platform:latest
 
       - name: Enter Inside Platform native container and run L2 Test
         run: |

test/functional-tests/features/usblogupload.feature

@@ -0,0 +1,44 @@
+Feature: USB Log Upload
+  This feature covers the USB log upload functionality, including error handling, archive creation, MAC address logging, temp directory cleanup, and success/failure scenarios.
+
+  Scenario: USB not mounted or missing log path
+    Given the USB log upload binary is available
+    When I run usblogupload with a non-existent mount point
+    Then the process should fail with code 2 or 3
+    And a failure message should be logged
+
+  Scenario: Archive creation on valid mount
+    Given a valid USB mount point
+    When I run usblogupload
+    Then the process should exit with code 0 or 3
+    And an archive creation log may appear
+
+  Scenario: MAC address and file log
+    Given a valid USB mount point
+    When I run usblogupload
+    Then the process should exit with code 0 or 3
+    And a log line with MAC address and file name may appear
+
+  Scenario: Temp directory cleanup
+    Given a valid USB mount point
+    When I run usblogupload
+    Then the process should exit with code 0 or 3
+    And a cleanup log may appear
+
+  Scenario: Successful USB log upload
+    Given the USB log upload binary is available
+    When I run usblogupload with a valid mount point
+    Then the process should exit with code 0
+    And a completion message should be logged
+
+  Scenario: Invalid usage
+    Given the USB log upload binary is available
+    When I run usblogupload with no arguments
+    Then the process should exit with code 4
+    And a log about failed logging system initialization may appear
+
+  Scenario: USB not mounted
+    Given the USB log upload binary is available
+    When I run usblogupload with an unmounted path
+    Then the process should exit with code 2
+    And a log about failed USB mount point validation may appear

test/functional-tests/tests/test_log_upload_onreboot_false_case.py

@@ -34,14 +34,9 @@ def test_upload_cron_present():
     sleep(20)
     assert "urn:settings:LogUploadSettings:UploadSchedule:cron" in grep_dcmdlogs("is present setting cron jobs")
 
-@pytest.mark.run(order=2)
-def test_upload_script_started_onboot_false():
-    assert "UploadOnReboot=0" in grep_dcmdlogs("Triggered uploadSTBLogs.sh with arguments")
-    assert "Called uploadLogOnReboot with false" in grep_dcmdlogs("Called uploadLogOnReboot with false")
-    sleep(420)
-
 @pytest.mark.run(order=3)
 def test_fw_cron_scheduled():
+    sleep(420)
     assert "Scheduling DCM_FW_UPDATE Job handle" in grep_dcmdlogs("Scheduling DCM_FW_UPDATE Job handle")
 
 
@@ -52,8 +47,7 @@ def test_upload_cron_scheduled():
 
 @pytest.mark.run(order=5)
 def test_upload_script_started():
-    assert "Start log upload Script"  in grep_dcmdlogs("Start log upload Script")
-    assert "Called uploadDCMLogs" in grep_dcmdlogs("Called uploadDCMLogs")
+    assert "Start log upload via library API"  in grep_dcmdlogs("Start log upload via library API")
 
 @pytest.mark.run(order=6)
 def test_fwupdate_script_started():

test/functional-tests/tests/test_log_upload_onreboot_true_case.py

@@ -28,11 +28,6 @@
 def test_upload_cron_present():
     assert "urn:settings:LogUploadSettings:UploadSchedule:cron" in grep_dcmdlogs("is present setting cron jobs")
 
-@pytest.mark.run(order=2)
-def test_upload_script_started():
-    assert "UploadOnReboot=1" in grep_dcmdlogs("Triggered uploadSTBLogs.sh with arguments")
-    assert "Called uploadLogOnReboot with true" in grep_dcmdlogs("Called uploadLogOnReboot with true")
-
 @pytest.mark.run(order=3)
 def test_fw_cron_scheduled():
     sleep(540)
@@ -50,7 +45,5 @@ def test_upload_cron_scheduled():
 
 @pytest.mark.run(order=6)
 def test_upload_started():
-    assert "Start log upload Script"  in grep_dcmdlogs("Start log upload Script")
-    assert "FLAG=0" in grep_dcmdlogs("Triggered uploadSTBLogs.sh with arguments")
-    assert "Called uploadDCMLogs" in grep_dcmdlogs("Called uploadDCMLogs")
+    assert "Start log upload via library API"  in grep_dcmdlogs("Start log upload via library API")
 

test/functional-tests/tests/test_uploadLogsNow.py

@@ -34,11 +34,9 @@
 
 def run_uploadlogsnow():
     """Execute uploadlogsnow using the specific binary command"""
-    cmd = "/usr/local/bin/logupload uploadlogsnow"
-    result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=300)
+    result = subprocess.run("/usr/local/bin/logupload uploadlogsnow >> /opt/logs/logupload.log.0",shell=True)
     return result
 
-
 class TestUploadLogsNow:
     """Test suite for uploadLogsNow immediate upload functionality"""
 

test/functional-tests/tests/test_uploadstblogs_normal_upload.py

@@ -54,29 +54,15 @@ def test_normal_upload_initialization(self):
 
         # Run uploadSTBLogs
         #result = run_uploadstblogs()
-
-        result = subprocess.run([
-        "/usr/local/bin/logupload",
-        "",
-        "1",
-        "1",
-        "true",
-        "HTTP",
-        "https://mockxconf:50058/"
-        ])
-
-
+        result = subprocess.run("/usr/local/bin/logupload '' 1 1 true HTTP https://mockxconf:50058/ >> /opt/logs/logupload.log.0",shell=True)
+
         # Verify initialization
         assert result.returncode == 0 or result.returncode == 1, "Upload process should complete"
 
         # Check initialization logs
         init_logs = grep_uploadstb_logs("Context initialization successful")
         assert len(init_logs) > 0, "Context should be initialized successfully"
 
-        # Verify device properties loaded
-        logs = grep_uploadstb_logs("DEVICE_TYPE")
-        assert len(logs) > 0, "Device type should be loaded from properties"
-
         collection_logs = grep_uploadstb_logs_regex(r"collect|archive|gather")
         assert len(collection_logs) > 0, "Log collection should be attempted"
 
@@ -109,16 +95,7 @@ def test_large_file_collection(self):
         """Test: Service collects large log files within limits"""
         # Create large test files (10MB each)
         large_files = create_large_test_log_files(count=3, size_mb=10)
-
-        result = subprocess.run([
-        "/usr/local/bin/logupload",
-        "",
-        "1",
-        "1",
-        "true",
-        "HTTP",
-        "https://mockxconf:50058/"
-        ])
+        result = subprocess.run("/usr/local/bin/logupload '' 1 1 true HTTP https://mockxconf:50058/ >> /opt/logs/logupload.log.0",shell=True)
 
         # Verify files were processed
 

test/functional-tests/tests/test_usb_logupload.py

@@ -0,0 +1,122 @@
+import subprocess
+import os
+import re
+import pytest
+
+USBLOGUPLOAD_BIN = "/usr/local/bin/usblogupload"
+LOG_FILE = "/opt/logs/logupload.log"  # Adjust if needed
+
+# Helper to grep logs
+
+def grep_usblogupload_logs(search: str):
+    search_result = []
+    search_pattern = re.compile(re.escape(search), re.IGNORECASE)
+    try:
+        with open(LOG_FILE, 'r', encoding='utf-8', errors='ignore') as file:
+            for line in file:
+                if search_pattern.search(line):
+                    search_result.append(line)
+    except Exception as e:
+        print(f"Could not read file {LOG_FILE}: {e}")
+    return search_result
+
+
+@pytest.fixture(autouse=True)
+def setup_device_properties(tmp_path):
+    # Path to device.properties for test
+    device_properties_path = os.path.join(os.path.dirname(__file__), "device.properties")
+    backup_path = device_properties_path + ".bak"
+    # Backup original if exists
+    if os.path.exists(device_properties_path):
+        os.rename(device_properties_path, backup_path)
+    # Ensure RDK_PROFILE=TV is present
+    with open(device_properties_path, "w", encoding="utf-8") as f:
+        f.write("RDK_PROFILE=TV\n")
+    yield
+    # Restore original after test
+    if os.path.exists(backup_path):
+        os.remove(device_properties_path)
+        os.rename(backup_path, device_properties_path)
+
+class TestUSBLogUpload:
+    def test_usblogupload_missing_log_path(self, tmp_path):
+        # Simulate missing log path by passing a non-existent mount point
+        usb_mount = str(tmp_path / "not_a_mount")
+        result = subprocess.run([USBLOGUPLOAD_BIN, usb_mount], capture_output=True, text=True)
+        with open(LOG_FILE, "a", encoding="utf-8") as f:
+            f.write(result.stdout)
+            f.write(result.stderr)
+        assert result.returncode == 2 or result.returncode == 3, "Should fail with USB not mounted or write error"
+        logs = grep_usblogupload_logs("Failed")
+        # Accept log file or process output containing 'fail', 'error', or 'not mounted'
+        output = (result.stdout + result.stderr).lower()
+        assert (
+            logs or
+            "fail" in output or
+            "error" in output or
+            "not mounted" in output
+        ), (
+            f"Should log a failure message. Got stdout: {result.stdout}, stderr: {result.stderr}"
+        )
+
+    def test_usblogupload_archive_creation(self, tmp_path):
+        # Simulate a valid mount and check for archive creation log
+        usb_mount = tmp_path / "usb"
+        usb_mount.mkdir()
+        result = subprocess.run([USBLOGUPLOAD_BIN, str(usb_mount)], capture_output=True, text=True)
+        with open(LOG_FILE, "a", encoding="utf-8") as f:
+            f.write(result.stdout)
+            f.write(result.stderr)
+        # Look for archive or compression log
+        logs = grep_usblogupload_logs("archive")
+        assert result.returncode in (0, 3), "Should exit with success or write error code"
+        # Archive log may or may not appear depending on implementation
+
+    def test_usblogupload_mac_address_log(self, tmp_path):
+        # Simulate a valid mount and check for MAC address log
+        usb_mount = tmp_path / "usb"
+        usb_mount.mkdir()
+        result = subprocess.run([USBLOGUPLOAD_BIN, str(usb_mount)], capture_output=True, text=True)
+        with open(LOG_FILE, "a", encoding="utf-8") as f:
+            f.write(result.stdout)
+            f.write(result.stderr)
+        logs = grep_usblogupload_logs(":.*File:")
+        # This checks for the log line with MAC address and file name
+        # (Regex match, may need adjustment based on actual log format)
+        assert result.returncode in (0, 3), "Should exit with success or write error code"
+
+    def test_usblogupload_temp_dir_cleanup(self, tmp_path):
+        # Simulate a valid mount and check for temp dir cleanup log
+        usb_mount = tmp_path / "usb"
+        usb_mount.mkdir()
+        result = subprocess.run([USBLOGUPLOAD_BIN, str(usb_mount)], capture_output=True, text=True)
+        with open(LOG_FILE, "a", encoding="utf-8") as f:
+            f.write(result.stdout)
+            f.write(result.stderr)
+        logs = grep_usblogupload_logs("cleanup")
+        # This checks for cleanup log line (if implemented)
+        assert result.returncode in (0, 3), "Should exit with success or write error code"
+    def test_usblogupload_success(self, tmp_path):
+        usb_mount = "/tmp"
+        # Run the binary and capture output
+        result = subprocess.run([USBLOGUPLOAD_BIN, usb_mount], capture_output=True, text=True)
+        # Write output to log file
+        with open(LOG_FILE, "a", encoding="utf-8") as f:
+            f.write(result.stdout)
+            f.write(result.stderr)
+        assert result.returncode == 0, "Should exit with success code 0"
+        # Check for expected log
+        logs = grep_usblogupload_logs("COMPLETED USB LOG UPLOAD")
+        assert logs, "Should log completion message"
+
+    def test_usblogupload_invalid_usage(self):
+        result = subprocess.run([USBLOGUPLOAD_BIN], capture_output=True)
+        assert result.returncode == 4, "Should exit with invalid usage code 4"
+        logs = grep_usblogupload_logs("Failed to initialize logging system")
+        # This log may or may not appear depending on implementation
+
+    def test_usblogupload_usb_not_mounted(self):
+        result = subprocess.run([USBLOGUPLOAD_BIN, "/tmp/notmounted"], capture_output=True)
+        assert result.returncode == 2, "Should exit with USB not mounted code 2"
+        logs = grep_usblogupload_logs("Failed to validate USB mount point")
+        # This log may or may not appear depending on implementation

test/run_uploadstblogs_l2.sh

@@ -31,6 +31,13 @@ mkdir -p "$RESULT_DIR"
 echo "LOG.RDK.DEFAULT" >> /etc/debug.ini
 
 # Ensure properties files exist
+
+if grep -q '^RDK_PROFILE=' /etc/device.properties; then
+  sed -i 's/^RDK_PROFILE=.*/RDK_PROFILE=TV/' /etc/device.properties
+else
+  echo 'RDK_PROFILE=TV' >> /etc/device.properties
+fi
+
 if ! grep -q "LOG_PATH=/opt/logs/" /etc/include.properties; then
     echo "LOG_PATH=/opt/logs/" >> /etc/include.properties
 fi
@@ -52,6 +59,8 @@ if ! grep -q "BUILD_TYPE=" /etc/device.properties; then
     echo "BUILD_TYPE=dev" >> /etc/device.properties
 fi
 
+echo "AA:BB:CC:dd:EE:FF" >> /tmp/.estb_mac
+
 cd /usr/common_utilities
 sed -i '/file_upload\.sslverify/s/= 1;/= 0;/' uploadutils/mtls_upload.c
 sed -i 's/\(ret_code = setCommonCurlOpt(curl, s3url, NULL, \)true\()\)/\1false\2/g' uploadutils/uploadUtil.c
@@ -72,44 +81,48 @@ echo "====================================="
 # Run test suites
 
 echo ""
-echo "1. Running UploadLogsNow Tests..."
+echo "1. Running usbLogupload Tests..."
 pytest -v --json-report --json-report-summary \
-    --json-report-file $RESULT_DIR/uploadLogsNow.json test/functional-tests/tests/test4.py
+    --json-report-file $RESULT_DIR/usb_logupload.json test/functional-tests/tests/test_usb_logupload.py
 
 echo ""
-echo "2. Running Error Handling Tests..."
+echo "2. Running UploadLogsNow Tests..."
+pytest -v --json-report --json-report-summary \
+    --json-report-file $RESULT_DIR/uploadLogsNow.json test/functional-tests/tests/test_uploadLogsNow.py
+
+echo ""
+echo "3. Running Error Handling Tests..."
 pytest -v --json-report --json-report-summary \
     --json-report-file $RESULT_DIR/error_handling.json test/functional-tests/tests/test_uploadstblogs_error_handling.py
 
-echo "AA:BB:CC:dd:EE:FF" >> /tmp/.estb_mac
 
 mkdir -p /opt/logs
 mkdir -p /opt/logs/PreviousLogs
 
 echo ""
-echo "3. Running Normal Upload Tests..."
+echo "4. Running Normal Upload Tests..."
 mkdir -p /opt/logs/PreviousLogs
 pytest -v --json-report --json-report-summary \
         --json-report-file $RESULT_DIR/upload_normal.json test/functional-tests/tests/test_uploadstblogs_normal_upload.py
 
 
 echo ""
-echo "4. Running Retry Logic Tests..."
+echo "5. Running Retry Logic Tests..."
 pytest -v --json-report --json-report-summary \
     --json-report-file $RESULT_DIR/retry_logic.json test/functional-tests/tests/test_uploadstblogs_retry_logic.py
 
 echo ""
-echo "5. Running Security Tests..."
+echo "6. Running Security Tests..."
 pytest -v --json-report --json-report-summary \
     --json-report-file $RESULT_DIR/security.json test/functional-tests/tests/test_uploadstblogs_security.py
 
 echo ""
-echo "6. Running Resource Management Tests..."
+echo "7. Running Resource Management Tests..."
 pytest -v --json-report --json-report-summary \
     --json-report-file $RESULT_DIR/resource_management.json test/functional-tests/tests/test_uploadstblogs_resource_management.py
 
 echo ""
-echo "7. Running Upload Strategy Tests..."
+echo "8. Running Upload Strategy Tests..."
 pytest -v --json-report --json-report-summary \
     --json-report-file $RESULT_DIR/upload_strategies.json test/functional-tests/tests/test_uploadstblogs_upload_strategies.py