Skip to content

Commit

Permalink
Add tests for CPE match database producer
Browse files Browse the repository at this point in the history
  • Loading branch information
timopollmeier committed Jan 15, 2025
1 parent 4a2cea9 commit 9243f78
Show file tree
Hide file tree
Showing 2 changed files with 432 additions and 0 deletions.
44 changes: 44 additions & 0 deletions tests/cpe_match/producer/mock_worker.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
# SPDX-FileCopyrightText: 2025 Greenbone AG
#
# SPDX-License-Identifier: GPL-3.0-or-later
from typing import Sequence

from pontos.nvd.models.cpe_match_string import CPEMatchString
from rich.console import Console
from rich.progress import Progress

from greenbone.scap.generic_cli.worker.base import BaseScapWorker


class CpeMatchMockWorker(BaseScapWorker[CPEMatchString]):

def __init__(
self,
console: Console,
error_console: Console,
progress: Progress,
*,
verbose: int | None = None,
):
super().__init__(
console=console,
error_console=error_console,
progress=progress,
verbose=verbose,
)
self.context_entered: bool = False
self.context_exited: bool = False
self.items_received: list[CPEMatchString] = []
self.item_count = 0
self.chunk_count = 0

async def _handle_chunk(self, chunk: Sequence[CPEMatchString]) -> None:
self.items_received.extend(chunk)
self.item_count += len(chunk)
self.chunk_count += 1

async def __aenter__(self):
self.context_entered = True

async def __aexit__(self, __exc_type, __exc_value, __traceback):
self.context_exited = True
Loading

0 comments on commit 9243f78

Please sign in to comment.