Add tests and Github workflows (#1)

.github/workflows/lint.yml

@@ -0,0 +1,39 @@
+name: lint
+
+on: [push]
+
+jobs:
+  black:
+    name: Check code formatting with black
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Extract black version from requirements-dev.txt
+        id: black_version
+        run: echo "::set-output name=version::$(grep -oP 'black==\K[^"]+' requirements-dev.txt)"
+      - uses: psf/black@stable
+        with:
+          src: ./src
+          version: ${{ steps.back_version.outputs.version }}
+  ruff:
+    name: Check code formatting with ruff
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Extract ruff version from requirements-dev.txt
+        id: ruff_version
+        run: echo "::set-output name=version::$(grep -oP 'ruff==\K[^"]+' requirements-dev.txt)"
+      - uses: chartboost/ruff-action@v1
+        with:
+          version: ${{ steps.ruff_version.outputs.version }}
+  mypy:
+    name: Check type hints with mypy
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v4
+        with:
+          python-version: '3.11'
+      - run: |
+          pip install -r requirements-dev.txt
+          mypy src/

.github/workflows/tests.yml

@@ -0,0 +1,23 @@
+name: tests
+
+on: [push]
+
+jobs:
+  test:
+    name: Run tests with pytest
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ['3.8', '3.9', '3.10', '3.11', '3.12']
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          pip install -r requirements-dev.txt
+      - name: Test with tox
+        run: tox

pyproject.toml

@@ -55,7 +55,10 @@ select = [
 ]
 line-length = 120
 src = ["src"]
-ignore = ["PLR0913"]
+ignore = ["E501", "PLR0913"]
 
 [tool.ruff.isort]
 known-first-party = ["clamdpy"]
+
+[tool.pytest.ini_options]
+pythonpath = "src"

requirements-dev.txt

@@ -0,0 +1,6 @@
+black==23.10.0
+mypy==1.6.1
+pytest==7.4.2
+ruff==0.1.1
+tox==4.11.3
+tox-gh-actions==3.1.3

src/clamdpy/exceptions.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+
 class ClamdException(Exception):
     pass
 

src/clamdpy/sockets.py

@@ -4,7 +4,7 @@
 import struct
 import sys
 from pathlib import Path
-from typing import ClassVar, Literal, overload
+from typing import Literal, overload
 
 from .exceptions import BufferTooLongError, CommandReadTimedOut, ConnectionError, ResponseError, UnknownCommand
 from .models import ScanResult, VersionInfo
@@ -18,8 +18,6 @@
 class ClamdNetworkSocket:
     """A class to interact with clamd with a network socket (`socket.AF_INET`)."""
 
-    socket_class: ClassVar[type[socket.socket]] = socket.socket
-
     def __init__(
         self,
         host: str = "127.0.0.1",
@@ -40,7 +38,7 @@ def _endline(self) -> Literal["\n", "\0"]:
 
     def _acquire_socket(self) -> socket.socket:
         try:
-            clamd_socket = self.socket_class(socket.AF_INET, socket.SOCK_STREAM)
+            clamd_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             clamd_socket.settimeout(self.timeout)
             clamd_socket.connect((self.host, self.port))
             return clamd_socket
@@ -49,17 +47,18 @@ def _acquire_socket(self) -> socket.socket:
                 raise ConnectionError(f"Error while connecting to {self.host}:{self.port}: {e.args[0]}")
             raise ConnectionError(f"Error while connecting to {self.host}:{self.port}: {e.args[1]}", int(e.args[0]))
 
-    def _command(self, command: str, *args: str, multiline: bool = True) -> str:
+    def _command(self, command: str, *args: str, multiline: bool = True, raise_on_error: bool = True) -> str:
         with self._acquire_socket() as sock:
             self._send(sock, command, *args)
             recv = self._recv(sock, multiline=multiline)
             if recv == UNKNOWN_COMMAND:
                 raise UnknownCommand(command)
             if recv == COMMAND_READ_TIMED_OUT:
                 raise CommandReadTimedOut(command)
-            response = recv.rsplit("ERROR", 1)
-            if len(response) > 1:
-                raise ResponseError(response[0])
+            if raise_on_error:
+                response = recv.rsplit("ERROR", 1)
+                if len(response) > 1:
+                    raise ResponseError(command, response[0].strip())
             return recv
 
     def _send(self, sock: socket.socket, command: str, *args: str) -> None:
@@ -92,7 +91,7 @@ def _any_scan(self, command: str, path: StrPath, raw: Literal[False] = ...) -> l
 
     def _any_scan(self, command: str, path: StrPath, raw: bool = False) -> list[ScanResult] | str:
         path = Path(path).absolute()
-        result = self._command(command, str(path))
+        result = self._command(command, str(path), raise_on_error=False)
         if raw:
             return result
         return [ScanResult._from_str(line, command) for line in result.split(self._endline)]
@@ -199,15 +198,15 @@ def instream(
         buff: SupportsRead[bytes],
         raw: Literal[False] = ...,
         max_chunk_size: int | None = ...,
-    ) -> list[ScanResult]:
+    ) -> ScanResult:
         ...
 
     def instream(
         self,
         buff: SupportsRead[bytes],
         raw: bool = False,
         max_chunk_size: int | None = None,
-    ) -> list[ScanResult] | str:
+    ) -> ScanResult | str:
         """Scan a stream of data. The stream is sent to clamd in chunks, after INSTREAM,
         on the same socket on which the command was sent.
         """
@@ -226,10 +225,10 @@ def instream(
 
             result = self._recv(sock)
             if "INSTREAM size limit exceeded" in result:
-                raise BufferTooLongError(result.rsplit("ERROR", 1)[0])
+                raise BufferTooLongError(result.rsplit("ERROR", 1)[0].strip())
             if raw:
                 return result
-            return [ScanResult._from_str(line, "INSTREAM", stream=True) for line in result.split(self._endline)]
+            return ScanResult._from_str(result, "INSTREAM", stream=True)
 
 
 class ClamdUnixSocket(ClamdNetworkSocket):
@@ -249,7 +248,7 @@ def __init__(
 
     def _acquire_socket(self) -> socket.socket:
         try:
-            clamd_socket = self.socket_class(socket.AF_UNIX, socket.SOCK_STREAM)
+            clamd_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
             clamd_socket.settimeout(self.timeout)
             clamd_socket.connect(str(self.socket_path))
             return clamd_socket

src/clamdpy/typing.py

@@ -11,4 +11,4 @@ def read(self, __length: int = ...) -> _T_co:
 
 
 # Taken from _typeshed/__init__.pyi
-StrPath = Union[str, PathLike[str]]
+StrPath = Union[str, "PathLike[str]"]