chore(BA-4041): Enable S101 linter rule and replace assert statements (#8364)

Co-authored-by: Claude Sonnet 4.5 <[email protected]>

Author: HyeockJinKim

changes/8364.enhance.md

@@ -0,0 +1 @@
+Enable S101 linter rule and replace assert statements

pyproject.toml

@@ -110,6 +110,7 @@ select = [
     "PLC0415", # pylint-convention: import-outside-toplevel - enforce imports at module level
     "ANN",    # flake8-annotations: enforces type annotations
     "DTZ",    # flake8-datetimez: enforces timezone-aware datetime usage
+    "S101",   # flake8-bandit: detects use of assert in production code
 ]
 ignore = [
     "E203",   # whitespace before ':' - conflicts with Black formatter
@@ -214,8 +215,8 @@ split-on-trailing-comma = true
 # - Client func vfolder has progress printing
 "src/ai/backend/client/func/vfolder.py" = ["T20"]
 # - Alembic migrations may print progress and have standard signatures
-"src/ai/backend/*/models/alembic/**/*.py" = ["T20", "ANN", "PLC0415"]
-"src/ai/backend/*/*/models/alembic/**/*.py" = ["T20", "ANN", "PLC0415"]
+"src/ai/backend/*/models/alembic/**/*.py" = ["T20", "ANN", "PLC0415", "S101"]
+"src/ai/backend/*/*/models/alembic/**/*.py" = ["T20", "ANN", "PLC0415", "S101"]
 # - Model modules may have circular dependencies between Row classes
 "src/ai/backend/*/models/**/*.py" = ["PLC0415"]
 "src/ai/backend/*/*/models/**/*.py" = ["PLC0415"]
@@ -255,8 +256,8 @@ split-on-trailing-comma = true
 "src/ai/backend/logging/logger.py" = ["PLC0415"]
 "src/ai/backend/logging/utils.py" = ["PLC0415"]
 # - Test files may use print for debugging and have flexible type annotations
-"tests/**/*.py" = ["T20", "ANN"]
-"src/ai/backend/test/**/*.py" = ["T20", "ANN"]
+"tests/**/*.py" = ["T20", "ANN", "S101"]
+"src/ai/backend/test/**/*.py" = ["T20", "ANN", "S101"]
 # - Type stub files have their own conventions
 "stubs/**/*.pyi" = ["ANN"]
 # - Vec2D class intentionally overrides tuple operators with incompatible return types for vector math

src/ai/backend/accelerator/hyperaccel/lpu/lpu_api.py

@@ -24,9 +24,10 @@ async def list_devices() -> list[LPUDevice]:
                 x for x in (ddp / "drm").iterdir() if x.name.startswith("renderD")
             ],
         )
-        assert len(renderD_files) == 1, (
-            f'no renderD file found under {display_driver_path.as_posix()}/"drm"!'
-        )
+        if len(renderD_files) != 1:
+            raise RuntimeError(
+                f'no renderD file found under {display_driver_path.as_posix()}/"drm"!'
+            )
         renderD_filename = renderD_files[0].name
         devices.append(
             LPUDevice(

src/ai/backend/accelerator/ipu/plugin.py

@@ -173,7 +173,8 @@ async def prepare_networks(self) -> None:
             if len(ipv4_subnet) > 1:
                 raise DockerNetworkError(f"Multiple IPv4 configs on network {network_identifier}")
             ip_network = ipaddress.ip_network(ipv4_subnet[0])
-            assert isinstance(ip_network, ipaddress.IPv4Network)
+            if not isinstance(ip_network, ipaddress.IPv4Network):
+                raise RuntimeError(f"Expected IPv4Network but got {type(ip_network).__name__}")
             self.subnet_network_map[ip_network] = network_identifier
 
     def get_docker_network(self, device: IPUDevice) -> str:

src/ai/backend/accelerator/mock/plugin.py

@@ -613,8 +613,8 @@ async def generate_docker_args(
         return docker_config
 
     def _get_share(self, device: MockDevice) -> Decimal:
-        if isinstance(device, CUDADevice):
-            assert not device.is_mig_device
+        if isinstance(device, CUDADevice) and device.is_mig_device:
+            raise ValueError("Cannot calculate share for MIG devices")
         return self._get_share_raw(device.memory_size, device.processing_units)
 
     def _get_share_raw(self, memory_size: int, subproc_count: int) -> Decimal:
@@ -642,8 +642,10 @@ async def generate_resource_data(
         active_device_id_list: list[DeviceId] = []
         for slot_type, per_device_alloc in device_alloc.items():
             if slot_type == SlotTypes.UNIQUE:
-                assert len(device_alloc) == 1
-                assert len(per_device_alloc) == 1
+                if len(device_alloc) != 1:
+                    raise ValueError("UNIQUE slot type requires exactly one slot allocation")
+                if len(per_device_alloc) != 1:
+                    raise ValueError("UNIQUE slot type requires exactly one device allocation")
                 device_id = list(per_device_alloc.keys())[0]
                 active_device_id_list.append(device_id)
                 is_unique = True
@@ -787,15 +789,17 @@ async def generate_mounts(
         return []
 
     def get_metadata(self) -> AcceleratorMetadata:
-        assert self._all_devices is not None
+        if self._all_devices is None:
+            raise RuntimeError("Devices not initialized. Call list_devices() first.")
         format_key = ""
         if self._mode == AllocationModes.DISCRETE:
             format_key = "device"
         else:
             format_key = "shares"
         if self.key == DeviceName("cuda"):
             for device in self._all_devices:
-                assert isinstance(device, CUDADevice)
+                if not isinstance(device, CUDADevice):
+                    raise RuntimeError(f"Expected CUDADevice but got {type(device).__name__}")
                 if device.is_mig_device:
                     format_key = "*-mig"
                     break

src/ai/backend/accelerator/rebellions/atom_max/plugin.py

@@ -45,9 +45,10 @@ async def _list_devices(self) -> list[ATOMMaxDevice]:
         for device_info in stats.devices:
             if device_info.name not in VALID_DEVICE_NAME:
                 continue
-            assert device_info.sid is not None, (
-                "sid value in atom-stats -j response cannot be null for ATOM Max device!"
-            )
+            if device_info.sid is None:
+                raise RuntimeError(
+                    "sid value in atom-stats -j response cannot be null for ATOM Max device!"
+                )
             devices_by_sid[DeviceId(device_info.sid)].append(
                 ATOMMaxChildDevice(
                     serial=device_info.uuid,

src/ai/backend/accelerator/rebellions/common/plugin.py

@@ -280,7 +280,8 @@ async def generate_docker_args(
         numa_node_indexes: set[int] = set()
         for dev in await self._list_devices():
             if dev.device_id in device_alloc.get(self.slot_types[0][0], {}).keys():
-                assert dev.numa_node is not None, "NUMA node index of accelerator cannot be null!"
+                if dev.numa_node is None:
+                    raise RuntimeError("NUMA node index of accelerator cannot be null!")
                 assigned_devices.append(dev)
                 device_files.extend([Path(x) for x in await self.list_device_files(dev)])
                 numa_node_indexes.add(dev.numa_node)

src/ai/backend/account_manager/config.py

@@ -83,9 +83,11 @@ def uid_validator(
         value: int | str | None,
     ) -> int:
         if value is None:
-            assert cls.default_uid, "value is None but default_uid not provided"
+            if not cls.default_uid:
+                raise ValueError("value is None but default_uid not provided")
             return cls.default_uid
-        assert isinstance(value, (int, str)), "value must be an integer"
+        if not isinstance(value, (int, str)):
+            raise TypeError("value must be an integer or string")
         match value:
             case int():
                 if value == -1:
@@ -153,8 +155,11 @@ def uid_validator(
         value: int | str | None,
     ) -> int:
         if value is None:
-            assert cls.default_gid, "value is None but default_gid not provided"
-        assert isinstance(value, (int, str)), "value must be an integer"
+            if not cls.default_gid:
+                raise ValueError("value is None but default_gid not provided")
+            return cls.default_gid
+        if not isinstance(value, (int, str)):
+            raise TypeError("value must be an integer or string")
         match value:
             case int():
                 if value == -1:
@@ -445,7 +450,8 @@ def generate_example_json(
         res = {}
         for name, info in schema.model_fields.items():
             config_key = [*parent, name]
-            assert info.annotation
+            if not info.annotation:
+                raise ValueError(f"Field {name} in {schema} has no annotation")
             alternative_example = Undefined
             if info.examples:
                 res[name] = info.examples[0]

src/ai/backend/account_manager/server.py

@@ -320,9 +320,8 @@ async def server_main(
     try:
         ssl_ctx = None
         if am_config.ssl_enabled:
-            assert am_config.ssl_cert is not None, (
-                "Should set `account_manager.ssl-cert` in config file."
-            )
+            if am_config.ssl_cert is None:
+                raise ValueError("Should set `account_manager.ssl-cert` in config file.")
             ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
             ssl_ctx.load_cert_chain(
                 str(am_config.ssl_cert),

src/ai/backend/cli/params.py

@@ -191,7 +191,8 @@ def convert(
         ctx: Optional[click.Context],
     ) -> Any:
         key, value = value.split("=", maxsplit=1)
-        assert self._rx_range_key.match(key), "The key must be a valid slug string."
+        if not self._rx_range_key.match(key):
+            self.fail("The key must be a valid slug string.", param, ctx)
         try:
             if value.startswith("case:"):
                 return key, value[5:].split(",")