feat: Use Gemini to set-up the basics of tests for our python cli

.github/workflows/python-tests.yml

@@ -0,0 +1,30 @@
+name: Python Tests
+
+on:
+  pull_request:
+    branches:
+      - main
+      - master
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.8", "3.9", "3.10"]
+
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          cd ./unix
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+          pip install -r dev-requirements.txt
+      - name: Test with pytest
+        run: |
+          pytest

.gitignore

@@ -1,2 +1,3 @@
 .DS_Store
 .vscode/
+*.pyc

unix/CONTRIBUTING.md

@@ -43,6 +43,22 @@ We use the following tools:
   src/machine_stats/*` or `pipenv run pylint src`)
 * `bump2version` for version bumps (`pipenv run bump2version`)
 
+### Testing
+
+We use `pytest` for testing. Test files are located in the `tests/` directory and should be named with a `test_` prefix (e.g., `test_my_feature.py`).
+
+To run the tests, use the following command:
+
+```console
+pytest
+```
+
+or
+
+```console
+pipenv run pytest
+```
+
 ### How to release a new Machine Stats version
 
 To deploy a new version of Machine Stats, you will need to create a release. The steps are pretty simple, You can find Github's instruction [here](https://docs.github.com/en/repositories/releasing-projects-on-github/managing-releases-in-a-repository#creating-a-release).

unix/Pipfile

@@ -18,6 +18,8 @@ twine = "*"
 wheel = "*"
 build = "*"
 bump2version = "*"
+pytest = "*"
+pytest-mock = "*"
 
 [scripts]
 stats = "machine_stats"

unix/dev-requirements.txt

@@ -63,3 +63,5 @@ webencodings==0.5.1
 wheel==0.38.4
 wrapt==1.14.1; python_version < '3.11'
 zipp==3.11.0; python_version < '3.10'
+pytest
+pytest-mock

unix/flake.nix

@@ -15,6 +15,8 @@
             # A Python interpreter including the 'venv' module is required to bootstrap
             # the environment.
             python3Packages.python
+            python3Packages.pytest
+            python3Packages.pluggy
 
             # This execute some shell code to initialize a venv in $venvDir before
             # dropping into the shell

unix/setup.py

@@ -5,7 +5,7 @@
 
 setup(
     name="machine_stats",
-    version="develop",
+    version="0.1.1",
     author="Tidal SW",
     author_email="support@tidalcloud.com",
     description="A simple and effective way to gather machine statistics (RAM, Storage, CPU, etc.) from server environment",

unix/src/machine_stats/modules/cpu_utilization.py

@@ -28,58 +28,79 @@ def run_module():
     module = AnsibleModule(argument_spec=module_args, supports_check_mode=True)
 
     # if timeout is specified to 0, that means that the module is disabled
-    if module.params["timeout"] == 0:
-        module.exit_json(**result)
-
+    # or
     # if the user is working with this module in only check mode we do not
     # want to make any changes to the environment, just return the current
     # state with no modifications
-    if module.check_mode:
-        module.exit_json(**result)
+    if module.params["timeout"] == 0 or module.check_mode:
+        return module.exit_json(**result)
 
     # get CPU utilization values
     try:
         if module.params["only_value"]:
             value, rtc_date, rtc_time = cpu_utilization_value(module.params["timeout"])
+            result["ansible_cpu_utilization"] = dict(
+                value=value, rtc_date=rtc_date, rtc_time=rtc_time
+            )
         else:
-            average, peak, rtc_date, rtc_time = cpu_utilization(module.params["timeout"])
+            average, peak, rtc_date, rtc_time = cpu_utilization(
+                module.params["timeout"]
+            )
+            result["ansible_cpu_utilization"] = dict(
+                average=average, peak=peak, rtc_date=rtc_date, rtc_time=rtc_time
+            )
     except Exception as e:
         module.fail_json(msg=str(e), **result)
 
     # manipulate or modify the state as needed
     result["timeout"] = module.params["timeout"]
-    if module.params["only_value"]:
-        result["ansible_cpu_utilization"] = dict(value=value, rtc_date=rtc_date, rtc_time=rtc_time)
-    else:
-        result["ansible_cpu_utilization"] = dict(average=average, peak=peak , rtc_date=rtc_date, rtc_time=rtc_time)
 
     module.exit_json(**result)
 
+
 def get_perf():
     with open("/proc/stat") as f:
         fields = [float(column) for column in f.readline().strip().split()[1:]]
         idle, total = fields[3], sum(fields)
         return idle, total
 
+
 def get_date_time():
     with open("/proc/driver/rtc") as t:
         rtc_time_line = t.readline().strip().split()
         rtc_date_line = t.readline().strip().split()
-    
+
         rtc_time = rtc_time_line[2]
         rtc_date = rtc_date_line[2]
-        
+
         return rtc_date, rtc_time
 
-def cpu_utilization(timeout):
+
+def cpu_utilization(timeout=1):
+    """
+    Calculate CPU utilization over a given timeout period.
+    :param timeout: Duration in seconds to monitor CPU utilization.
+    :return: Tuple containing average utilization, peak utilization,
+                RTC date, and RTC time.
+
+    The default value is 1, which causes the function to run 1 time.
+    So if timeout is less than 1, return zeros.
+
+    """
     last_idle = last_total = total_runs = 0
     cpu_stats = []
 
+    if timeout < 1:
+        return 0, 0, 0, 0
+
     while total_runs < timeout:
         idle, total = get_perf()
         idle_delta, total_delta = idle - last_idle, total - last_total
         last_idle, last_total = idle, total
-        utilisation = 100.0 * (1.0 - idle_delta / total_delta)
+        if total_delta == 0:
+            utilisation = 0.0
+        else:
+            utilisation = 100.0 * (1.0 - idle_delta / total_delta)
         cpu_stats.append(utilisation)
         total_runs += 1
         sleep(1)
@@ -93,13 +114,21 @@ def cpu_utilization(timeout):
 
 
 def cpu_utilization_value(timeout):
+    """
+    Calculate CPU utilization over a given timeout period.
+    :param timeout: Duration in seconds to monitor CPU utilization.
+    :return: Tuple containing utilization, RTC date, and RTC time.
+    """
     last_idle, last_total = get_perf()
     rtc_date, rtc_time = get_date_time()
 
     sleep(timeout)
     idle, total = get_perf()
     idle_delta, total_delta = idle - last_idle, total - last_total
-    utilization = 100.0 * (1.0 - idle_delta / total_delta)
+    if total_delta == 0:
+        utilization = 0.0
+    else:
+        utilization = 100.0 * (1.0 - idle_delta / total_delta)
 
     return utilization, rtc_date, rtc_time
 

unix/src/machine_stats/modules/proc_stats.py

@@ -25,50 +25,83 @@ def run_module():
     # this includes instantiation, a couple of common attr would be the
     # args/params passed to the execution, as well as if the module
     # supports check mode
-    module = AnsibleModule(argument_spec=module_args, supports_check_mode=False)
+    module = AnsibleModule(argument_spec=module_args, supports_check_mode=True)
 
-    # if proc-stats isn't defined or false that means that the module
-    # is disabled
-    if not module.params["process_stats"]:
+    # if proc-stats is disabled or we are in check mode, exit early.
+    if not module.params["process_stats"] or module.check_mode:
         module.exit_json(**result)
-
-    # if the user is working with this module in only check mode we do not
-    # want to make any changes to the environment, just return the current
-    # state with no modifications
-    if module.check_mode:
-        module.exit_json(**result)
-
-    # get stats from processes running on server
+    else:
+        # get stats from processes running on server
+        try:
+            stats = process_stats()
+            result["ansible_proc_stats"] = stats
+            module.exit_json(**result)
+        except Exception as e:
+            module.fail_json(msg=str(e), **result)
+
+
+def _get_process_exe_info(process_path):
+    """
+    Resolves the executable path for a process to get its path and name.
+    Returns a tuple of (path, name) or None on failure.
+    """
     try:
-        stats = process_stats()
-    except Exception as e:
-        module.fail_json(msg=str(e), **result)
+        path, name = str(Path(process_path + "/exe").resolve()).rsplit("/", 1)
+        return path, name
+    except Exception:
+        return None
 
-    result["ansible_proc_stats"] = stats
 
-    module.exit_json(**result)
+def _parse_proc_status_file(process_path):
+    """
+    Parses the /proc/<pid>/status file and returns a dictionary of its contents.
+    """
+    # This algorithm expects all status files to be formatted like:
+    # ```
+    # Name:	init
+    # Umask:	0000
+    # State:	S (sleeping)
+    # Tgid:	1
+    # Ngid:	0
+    # Pid:	1
+    # PPid:	0
+    # ...
+    # ```
+    status = {}
+    with open(process_path + "/status") as proc_status:
+        for line in proc_status:
+            result = line.split(":")
+            key = result[0].lower().strip()
+            value = result[1].strip()
+            status[key] = value
+    return status
+
+
+def _get_username_from_uid(uid_str):
+    """
+    Takes a UID string from a proc status file and returns the username.
+    Falls back to the UID if the username cannot be found.
+    """
+    uid = int(uid_str.split()[0])
+    try:
+        user_entry = getpwuid(uid)
+        return user_entry.pw_name
+    except KeyError:
+        return str(uid)  # Fallback to UID
 
 
 def parse_status(process_path):
     """Takes in a string value of a process path, returns a dictionary containing the
     following attributes ['path', 'name', 'total_alive_time', 'pid', 'ppid',
     'max_memory_used_mb', 'memory_used_mb']"""
 
-    stats = dict()
-    # Follow the symlink for the exe file to get the path and name of
-    # the executable
-    #
-    # Note : This call requires root level access to be able to follow
-    # all symlinks. Otherwise some processes will be identified as
-    # /proc/2138/exe
+    stats = {}
 
-    try:
-        path, name = str(Path(process_path + "/exe").resolve()).rsplit("/", 1)
-    except:
+    exe_info = _get_process_exe_info(process_path)
+    if not exe_info:
         return stats
 
-    stats["path"] = path
-    stats["name"] = name
+    stats["path"], stats["name"] = exe_info
 
     # Use the folder create time for the process as an indicator for
     # the start time.
@@ -78,23 +111,7 @@ def parse_status(process_path):
     # that the folder metadata has not changed since creation
     stats["total_alive_time"] = round(time() - os.stat(process_path).st_ctime)
 
-    # This algorithm expects all status files to be formatted like:
-    # ```
-    # Name:	init
-    # Umask:	0000
-    # State:	S (sleeping)
-    # Tgid:	1
-    # Ngid:	0
-    # Pid:	1
-    # PPid:	0
-    # ...
-    # ```
-    status = dict()
-    with open(process_path + "/status") as proc_status:
-        for line in proc_status:
-            result = line.split(":")
-            name = result[0].lower().strip()
-            status[name] = result[1].strip()
+    status = _parse_proc_status_file(process_path)
 
     # Parse error will throw to parent function
     if status.get("pid"):
@@ -125,15 +142,7 @@ def parse_status(process_path):
 
     # Finally, let's see if we can read the username
     if status.get("uid"):
-        uid = int(status["uid"].split()[0])
-        try:
-            user_entry = getpwuid(
-                uid
-            )  # KeyError is raised if the entry asked for cannot be found.
-        except KeyError:
-            stats["user"] = str(uid)  # Fallback to UID
-        else:
-            stats["user"] = user_entry.pw_name
+        stats["user"] = _get_username_from_uid(status["uid"])
 
     return stats