diff --git a/.github/workflows/python-ci.yml b/.github/workflows/python-ci.yml index d94db348..516315f7 100644 --- a/.github/workflows/python-ci.yml +++ b/.github/workflows/python-ci.yml @@ -37,10 +37,11 @@ jobs: run: | poetry install - - name: Lint with flake8 + - name: Lint with ruff if: startsWith(matrix.os, 'ubuntu') run: | - poetry run flake8 --max-line-length=127 + poetry run ruff check . + poetry run ruff format --check . - name: Type checking with mypy if: startsWith(matrix.os, 'ubuntu') diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 49414e64..fbc8a34d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -15,22 +15,18 @@ repos: - id: end-of-file-fixer - id: mixed-line-ending - id: trailing-whitespace - - repo: https://github.com/asottile/pyupgrade - rev: v3.7.0 - hooks: - - id: pyupgrade - args: [--py38-plus] - repo: local hooks: - id: system - name: black - entry: poetry run black . + name: ruff + entry: poetry run ruff check . --fix pass_filenames: false language: system types: [python] + require_serial: true - id: system - name: isort - entry: poetry run isort . + name: ruff-format + entry: poetry run ruff format . pass_filenames: false language: system types: [python] @@ -40,14 +36,6 @@ repos: pass_filenames: false language: system types: [python] - - id: system - name: flake8 - entry: poetry run flake8 . - pass_filenames: false - language: system - types: [python] - require_serial: true - args: [--darglint-ignore-regex, .*] # Skip docstring checks - id: system name: mypy entry: poetry run mypy censys @@ -55,9 +43,3 @@ repos: language: system types: [python] require_serial: true - # - id: system - # name: update autocomplete - # entry: bash scripts/update_autocomplete.sh - # pass_filenames: false - # language: system - # types: [] diff --git a/censys/asm/api.py b/censys/asm/api.py index a47cce67..e4348354 100644 --- a/censys/asm/api.py +++ b/censys/asm/api.py @@ -1,8 +1,9 @@ """Base for interacting with the Censys ASM API.""" import os +from collections.abc import Iterator from math import inf -from typing import Iterator, Optional, Type +from typing import Optional from requests.models import Response @@ -38,25 +39,17 @@ def __init__(self, api_key: Optional[str] = None, **kwargs): config = get_config() # Try to get credentials - self._api_key = ( - api_key - or os.getenv("CENSYS_ASM_API_KEY") - or config.get(DEFAULT, "asm_api_key") - ) + self._api_key = api_key or os.getenv("CENSYS_ASM_API_KEY") or config.get(DEFAULT, "asm_api_key") if not self._api_key: raise CensysException("No ASM API key configured.") - self._session.headers.update( - {"Content-Type": "application/json", "Censys-Api-Key": self._api_key} - ) + self._session.headers.update({"Content-Type": "application/json", "Censys-Api-Key": self._api_key}) def _get_exception_class( # type: ignore self, res: Response - ) -> Type[CensysAsmException]: - return CensysExceptionMapper.ASM_EXCEPTIONS.get( - res.json().get("errorCode"), CensysAsmException - ) + ) -> type[CensysAsmException]: + return CensysExceptionMapper.ASM_EXCEPTIONS.get(res.json().get("errorCode"), CensysAsmException) def _get_page( self, @@ -91,9 +84,7 @@ def _get_page( yield from res[keyword] - def _get_logbook_page( - self, path: str, args: Optional[dict] = None - ) -> Iterator[dict]: + def _get_logbook_page(self, path: str, args: Optional[dict] = None) -> Iterator[dict]: """Fetches paginated ASM logbook API events. Args: diff --git a/censys/asm/assets/assets.py b/censys/asm/assets/assets.py index 3baeb75b..25c8d588 100644 --- a/censys/asm/assets/assets.py +++ b/censys/asm/assets/assets.py @@ -1,11 +1,13 @@ """Base for interacting with the Censys Assets API.""" import re -from typing import Any, Dict, Iterator, List, Optional +from collections.abc import Iterator +from typing import Any, Optional -from ..api import CensysAsmAPI from censys.common.exceptions import CensysInvalidColorException +from ..api import CensysAsmAPI + HEX_REGEX = re.compile(r"^#(?:[0-9a-fA-F]{3}){1,2}$") @@ -42,9 +44,9 @@ def get_assets( self, page_number: int = 1, page_size: Optional[int] = None, - tag: Optional[List[str]] = None, + tag: Optional[list[str]] = None, tag_operator: Optional[str] = None, - source: Optional[List[str]] = None, + source: Optional[list[str]] = None, discovery_trail: Optional[bool] = None, ) -> Iterator[dict]: """Requests assets data. @@ -60,7 +62,7 @@ def get_assets( Yields: dict: The assets result returned. """ - args: Dict[str, Any] = {} + args: dict[str, Any] = {} if tag: args["tag"] = tag if tag_operator: @@ -69,9 +71,7 @@ def get_assets( args["source"] = source if discovery_trail: args["discoveryTrail"] = discovery_trail - yield from self._get_page( - self.base_path, page_number=page_number, page_size=page_size, args=args - ) + yield from self._get_page(self.base_path, page_number=page_number, page_size=page_size, args=args) def get_asset_by_id(self, asset_id: str) -> dict: """Requests asset data by ID. @@ -104,9 +104,7 @@ def get_comments( """ path = f"{self.base_path}/{self._format_asset_id(asset_id)}/comments" - return self._get_page( - path, page_number=page_number, page_size=page_size, keyword="comments" - ) + return self._get_page(path, page_number=page_number, page_size=page_size, keyword="comments") def get_comment_by_id(self, asset_id: str, comment_id: int) -> dict: """Requests a comment on a specified asset by comment ID. @@ -118,9 +116,7 @@ def get_comment_by_id(self, asset_id: str, comment_id: int) -> dict: Returns: dict: Comment search result. """ - path = ( - f"{self.base_path}/{self._format_asset_id(asset_id)}/comments/{comment_id}" - ) + path = f"{self.base_path}/{self._format_asset_id(asset_id)}/comments/{comment_id}" return self._get(path) @@ -149,9 +145,7 @@ def delete_comment(self, asset_id: str, comment_id: int) -> dict: Returns: dict: Deleted comment results. """ - path = ( - f"{self.base_path}/{self._format_asset_id(asset_id)}/comments/{comment_id}" - ) + path = f"{self.base_path}/{self._format_asset_id(asset_id)}/comments/{comment_id}" return self._delete(path) diff --git a/censys/asm/assets/domains.py b/censys/asm/assets/domains.py index 0645ea67..a148c773 100644 --- a/censys/asm/assets/domains.py +++ b/censys/asm/assets/domains.py @@ -1,6 +1,7 @@ """Interact with the Censys Domain Assets API.""" -from typing import Iterator, Optional +from collections.abc import Iterator +from typing import Optional from .assets import Assets @@ -17,9 +18,7 @@ def __init__(self, *args, **kwargs): """ super().__init__("domains", *args, **kwargs) - def get_subdomains( - self, domain: str, page_number: int = 1, page_size: Optional[int] = None - ) -> Iterator[dict]: + def get_subdomains(self, domain: str, page_number: int = 1, page_size: Optional[int] = None) -> Iterator[dict]: """List all subdomains of the parent domain. Args: diff --git a/censys/asm/assets/subdomains.py b/censys/asm/assets/subdomains.py index 759b243c..6fefa809 100644 --- a/censys/asm/assets/subdomains.py +++ b/censys/asm/assets/subdomains.py @@ -1,6 +1,7 @@ """Interact with the Censys Subdomain Assets API.""" -from typing import Any, Dict, Iterator, List, Optional +from collections.abc import Iterator +from typing import Any, Optional from .assets import Assets @@ -21,9 +22,9 @@ def get_assets( self, page_number: int = 1, page_size: Optional[int] = None, - tag: Optional[List[str]] = None, + tag: Optional[list[str]] = None, tag_operator: Optional[str] = None, - source: Optional[List[str]] = None, + source: Optional[list[str]] = None, discovery_trail: Optional[bool] = None, ) -> Iterator[dict]: """Requests assets data. @@ -41,7 +42,7 @@ def get_assets( Yields: dict: The assets result returned. """ - args: Dict[str, Any] = {} + args: dict[str, Any] = {} if tag: args["tag"] = tag if tag_operator: diff --git a/censys/asm/assets/web_entities.py b/censys/asm/assets/web_entities.py index 6c994164..3653435a 100644 --- a/censys/asm/assets/web_entities.py +++ b/censys/asm/assets/web_entities.py @@ -1,6 +1,7 @@ """Interact with the Censys Web Entities Assets API.""" -from typing import Iterator, Optional +from collections.abc import Iterator +from typing import Optional from .assets import Assets diff --git a/censys/asm/beta.py b/censys/asm/beta.py index cce578d4..dbf06db4 100644 --- a/censys/asm/beta.py +++ b/censys/asm/beta.py @@ -1,6 +1,6 @@ """Interact with miscellaneous Censys Beta APIs.""" -from typing import List, Optional +from typing import Optional from ..common.types import Datetime from ..common.utils import format_iso8601 @@ -12,9 +12,7 @@ class Beta(CensysAsmAPI): base_path = "/beta" - def get_logbook_data( - self, filters: Optional[dict] = None, cursor: Optional[str] = None - ): + def get_logbook_data(self, filters: Optional[dict] = None, cursor: Optional[str] = None): """Retrieve logbook data. Args: @@ -30,7 +28,7 @@ def get_logbook_data( def add_cloud_assets( self, cloud_connector_uid: str, - cloud_assets: List[dict], + cloud_assets: list[dict], ): """Add cloud assets. diff --git a/censys/asm/inventory.py b/censys/asm/inventory.py index 1226ff11..b23dc1f5 100644 --- a/censys/asm/inventory.py +++ b/censys/asm/inventory.py @@ -1,7 +1,7 @@ """Interact with the Censys Inventory Search API.""" import warnings -from typing import List, Optional +from typing import Optional from .api import CensysAsmAPI @@ -13,12 +13,12 @@ class InventorySearch(CensysAsmAPI): def search( self, - workspaces: Optional[List[str]] = None, + workspaces: Optional[list[str]] = None, query: Optional[str] = None, page_size: Optional[int] = None, cursor: Optional[str] = None, - sort: Optional[List[str]] = None, - fields: Optional[List[str]] = None, + sort: Optional[list[str]] = None, + fields: Optional[list[str]] = None, pages: Optional[int] = None, ) -> dict: """Search inventory data. @@ -75,10 +75,7 @@ def search( while next_cursor and (pages == -1 or page < pages): args["cursor"] = next_cursor resp = self._get(self.base_path, args=args) - if "nextCursor" in resp: - next_cursor = resp.get("nextCursor") - else: - next_cursor = None + next_cursor = resp.get("nextCursor") if "nextCursor" in resp else None hits.extend(resp.get("hits", [])) page += 1 @@ -87,7 +84,7 @@ def search( def aggregate( self, - workspaces: List[str], + workspaces: list[str], query: Optional[str] = None, aggregation: Optional[dict] = None, ) -> dict: @@ -109,7 +106,7 @@ def aggregate( return self._post(f"{self.base_path}/aggregate", data=body) - def fields(self, fields: Optional[List[str]] = None) -> dict: + def fields(self, fields: Optional[list[str]] = None) -> dict: """List inventory fields. If no fields are specified, all fields will be returned. diff --git a/censys/asm/logbook.py b/censys/asm/logbook.py index 69ea96ee..d5f16bfc 100644 --- a/censys/asm/logbook.py +++ b/censys/asm/logbook.py @@ -1,7 +1,8 @@ """Interact with the Censys Logbook API.""" import datetime -from typing import Iterator, List, Optional, Union +from collections.abc import Iterator +from typing import Optional, Union from .api import CensysAsmAPI @@ -14,7 +15,7 @@ class Logbook(CensysAsmAPI): def get_cursor( self, start: Optional[Union[datetime.datetime, int]] = None, - filters: Optional[List[str]] = None, + filters: Optional[list[str]] = None, ) -> str: """Requests a logbook cursor. @@ -71,7 +72,7 @@ class Filters: def format_data( start: Optional[Union[datetime.datetime, int]] = None, - filters: Optional[List[str]] = None, + filters: Optional[list[str]] = None, ) -> dict: """Formats cursor request data into a start date/id and filter list. diff --git a/censys/asm/risks.py b/censys/asm/risks.py index cb042d3c..511c7c14 100644 --- a/censys/asm/risks.py +++ b/censys/asm/risks.py @@ -1,7 +1,7 @@ """Interact with the Censys Risks API.""" import urllib.parse -from typing import Any, Dict, List, Optional +from typing import Any, Optional from .api import CensysAsmAPI @@ -36,7 +36,7 @@ def get_risk_events( Returns: dict: Risk events result. """ - args: Dict[str, Any] = {} + args: dict[str, Any] = {} if start: args["start"] = start if end: @@ -53,9 +53,7 @@ def get_risk_events( headers={"Accept": accept} if accept else None, ) - def get_risk_instances( - self, include_events: Optional[bool] = None, accept: Optional[str] = None - ) -> dict: + def get_risk_instances(self, include_events: Optional[bool] = None, accept: Optional[str] = None) -> dict: """Retrieve risk instances. Args: @@ -99,9 +97,7 @@ def search_risk_instances(self, data: dict, accept: Optional[str] = None) -> dic headers={"Accept": accept} if accept else None, ) - def get_risk_instance( - self, risk_instance_id: int, include_events: Optional[bool] = None - ) -> dict: + def get_risk_instance(self, risk_instance_id: int, include_events: Optional[bool] = None) -> dict: """Retrieve a risk instance. Args: @@ -130,7 +126,7 @@ def get_risk_types( self, limit: Optional[int] = None, page: Optional[int] = None, - sort: Optional[List[str]] = None, + sort: Optional[list[str]] = None, include_events: Optional[bool] = None, accept: Optional[str] = None, ) -> dict: @@ -146,7 +142,7 @@ def get_risk_types( Returns: dict: Risk types result. """ - args: Dict[str, Any] = {"sort": sort, "includeEvents": include_events} + args: dict[str, Any] = {"sort": sort, "includeEvents": include_events} if page: args["page"] = page if limit: @@ -157,9 +153,7 @@ def get_risk_types( headers={"Accept": accept} if accept else None, ) - def get_risk_type( - self, risk_type: str, include_events: Optional[bool] = None - ) -> dict: + def get_risk_type(self, risk_type: str, include_events: Optional[bool] = None) -> dict: """Retrieve a risk type. Args: diff --git a/censys/asm/seeds.py b/censys/asm/seeds.py index 411198fd..abab6480 100644 --- a/censys/asm/seeds.py +++ b/censys/asm/seeds.py @@ -1,6 +1,6 @@ """Interact with the Censys Seeds API.""" -from typing import List, Optional +from typing import Optional from .api import CensysAsmAPI @@ -12,9 +12,7 @@ class Seeds(CensysAsmAPI): base_path = "/v1/seeds" - def get_seeds( - self, seed_type: Optional[str] = None, label: Optional[str] = None - ) -> List[dict]: + def get_seeds(self, seed_type: Optional[str] = None, label: Optional[str] = None) -> list[dict]: """Requests seed data. Args: @@ -60,9 +58,7 @@ def add_seeds(self, seeds: list, force: Optional[bool] = None) -> dict: return self._post(self.base_path, args=args, data=data) - def replace_seeds_by_label( - self, label: str, seeds: list, force: Optional[bool] = None - ) -> dict: + def replace_seeds_by_label(self, label: str, seeds: list, force: Optional[bool] = None) -> dict: """Replace seeds in the ASM platform by label. Args: diff --git a/censys/cli/__init__.py b/censys/cli/__init__.py index d3d9ed7b..1d79fc4c 100644 --- a/censys/cli/__init__.py +++ b/censys/cli/__init__.py @@ -1,13 +1,15 @@ #!/usr/bin/env python3 # PYTHON_ARGCOMPLETE_OK """Interact with the Censys Search API through the command line.""" + import sys import argcomplete -from .args import get_parser from censys.common.version import __version__ +from .args import get_parser + def main(): """Main cli function.""" diff --git a/censys/cli/args.py b/censys/cli/args.py index 8bde77d4..eebbfa28 100644 --- a/censys/cli/args.py +++ b/censys/cli/args.py @@ -3,9 +3,10 @@ import argparse import os -from . import commands from censys.common.config import DEFAULT, get_config +from . import commands + def get_parser() -> argparse.ArgumentParser: """Gets ArgumentParser for CLI. diff --git a/censys/cli/commands/account.py b/censys/cli/commands/account.py index 231ec1ed..efaca684 100644 --- a/censys/cli/commands/account.py +++ b/censys/cli/commands/account.py @@ -23,9 +23,7 @@ def cli_account(args: argparse.Namespace): # pragma: no cover if args.json: console.print_json(data=account) else: - table = Table( - "Key", "Value", show_header=False, box=box.SQUARE, highlight=True - ) + table = Table("Key", "Value", show_header=False, box=box.SQUARE, highlight=True) table.add_row("Email", account["email"]) table.add_row("Login ID", account["login"]) table.add_row("First Login", account["first_login"]) @@ -33,7 +31,7 @@ def cli_account(args: argparse.Namespace): # pragma: no cover quota = account["quota"] table.add_row( "Query Quota", - f"{quota['used']} / {quota['allowance']} ({quota['used']/quota['allowance'] * 100 :.2f}%)", # noqa + f"{quota['used']} / {quota['allowance']} ({quota['used'] / quota['allowance'] * 100:.2f}%)", # noqa ) table.add_row("Quota Resets At", quota["resets_at"]) console.print(table) @@ -56,7 +54,5 @@ def include(parent_parser: argparse._SubParsersAction, parents: dict): help="check Censys account details and quota", parents=[parents["auth"]], ) - account_parser.add_argument( - "-j", "--json", action="store_true", help="Output in JSON format" - ) + account_parser.add_argument("-j", "--json", action="store_true", help="Output in JSON format") account_parser.set_defaults(func=cli_account) diff --git a/censys/cli/commands/asm.py b/censys/cli/commands/asm.py index 0dd94d5b..47315d92 100644 --- a/censys/cli/commands/asm.py +++ b/censys/cli/commands/asm.py @@ -6,7 +6,7 @@ import json import sys import threading -from typing import Dict, List, Union +from typing import Union from xml.etree import ElementTree from rich.progress import Progress, TaskID @@ -55,9 +55,7 @@ def cli_asm_config(_: argparse.Namespace): # pragma: no cover console.print("Please enter valid credentials") sys.exit(1) - color = Confirm.ask( - "Do you want color output?", default=True, show_default=False, console=console - ) + color = Confirm.ask("Do you want color output?", default=True, show_default=False, console=console) config.set(DEFAULT, "color", "auto" if color else "") try: @@ -72,7 +70,7 @@ def cli_asm_config(_: argparse.Namespace): # pragma: no cover sys.exit(1) -def get_seeds_from_xml(file: str) -> List[Dict[str, str]]: +def get_seeds_from_xml(file: str) -> list[dict[str, str]]: """Get seeds from nmap xml. Args: @@ -106,9 +104,7 @@ def get_seeds_from_xml(file: str) -> List[Dict[str, str]]: ] -def get_seeds_from_params( - args: argparse.Namespace, command_name: str -) -> List[Dict[str, Union[str, int]]]: +def get_seeds_from_params(args: argparse.Namespace, command_name: str) -> list[dict[str, Union[str, int]]]: """Get seeds from params. Args: @@ -134,9 +130,7 @@ def get_seeds_from_params( csv_reader = csv.DictReader(file, delimiter=",") if csv_reader.fieldnames: # Lowercase the field names - csv_reader.fieldnames = [ - string.lower() for string in csv_reader.fieldnames - ] + csv_reader.fieldnames = [string.lower() for string in csv_reader.fieldnames] for row in csv_reader: seeds.append(row) else: @@ -169,11 +163,7 @@ def get_seeds_from_params( if command_name != "delete-seeds" and "type" not in seed: seed["type"] = args.default_type - if ( - command_name != "replace-labeled-seeds" - and "label" not in seed - and "label" in args - ): + if command_name != "replace-labeled-seeds" and "label" not in seed and "label" in args: seed["label"] = args.label # The back end is really picky about sending extra fields, so we'll prune out anything it won't like. @@ -219,9 +209,7 @@ def cli_add_seeds(args: argparse.Namespace): if added_count < to_add_count: console.print(f"Seeds not added: {to_add_count - added_count}") if args.verbose: # pragma: no cover - console.print( - "The following seed(s) were not able to be added as they already exist or are reserved." - ) + console.print("The following seed(s) were not able to be added as they already exist or are reserved.") for seed in seeds_to_add: if not any(s for s in added_seeds if seed["value"] == s["value"]): console.print(f"{seed}") @@ -282,9 +270,7 @@ def delete_seed(seed_id: int, progress: Progress, task_id: TaskID): # Create a rich Progress instance with Progress() as progress: - progress_task_id = progress.add_task( - "[cyan]Deleting[/cyan]", total=len(seeds_to_delete) - ) + progress_task_id = progress.add_task("[cyan]Deleting[/cyan]", total=len(seeds_to_delete)) tasks = [] with concurrent.futures.ThreadPoolExecutor(max_workers=25) as executor: # Submit requests using the executor @@ -300,9 +286,7 @@ def delete_seed(seed_id: int, progress: Progress, task_id: TaskID): console.print(f"Deleted {len(seed_ids_deleted)} seeds.") if len(seed_ids_not_found) > 0: - console.print( - f"Unable to delete {len(seed_ids_not_found)} seeds because they were not present." - ) + console.print(f"Unable to delete {len(seed_ids_not_found)} seeds because they were not present.") def cli_delete_all_seeds(args: argparse.Namespace): @@ -346,9 +330,7 @@ def delete_seed(seed_id: int, progress: Progress, task_id: TaskID): with concurrent.futures.ThreadPoolExecutor(max_workers=25) as executor: # Submit requests using the executor for seed in seeds: - task = executor.submit( - delete_seed, seed["id"], progress, progress_task_id - ) + task = executor.submit(delete_seed, seed["id"], progress, progress_task_id) tasks.append(task) # Wait for all requests to complete @@ -395,9 +377,7 @@ def cli_replace_seeds_with_label(args: argparse.Namespace): console.print(f" {removed_seed}") skipped_seeds = res.get("skippedReservedSeeds", []) if len(skipped_seeds) > 0: - console.print( - "The following seed(s) were not added because they are reserved." - ) + console.print("The following seed(s) were not added because they are reserved.") for skipped_seed in skipped_seeds: console.print(f" {skipped_seed}") @@ -442,12 +422,8 @@ def add_seed_arguments(parser: argparse._SubParsersAction, is_delete=False) -> N help="input file name containing valid seeds in JSON format, unless --csv is specified (use - for stdin)", type=str, ) - seeds_group.add_argument( - "--json", "-j", help="input string containing valid json seeds", type=str - ) - seeds_group.add_argument( - "--nmap-xml", help="input file name containing valid xml nmap output", type=str - ) + seeds_group.add_argument("--json", "-j", help="input string containing valid json seeds", type=str) + seeds_group.add_argument("--nmap-xml", help="input file name containing valid xml nmap output", type=str) def cli_list_saved_queries(args: argparse.Namespace): @@ -458,9 +434,7 @@ def cli_list_saved_queries(args: argparse.Namespace): """ s = SavedQueries(args.api_key) try: - res = s.get_saved_queries( - args.query_name_prefix, args.page_size, args.page, args.filter_term - ) + res = s.get_saved_queries(args.query_name_prefix, args.page_size, args.page, args.filter_term) if args.csv: console.print("queryId,queryName,query,createdAt") @@ -549,9 +523,7 @@ def cli_execute_saved_query_by_name(args: argparse.Namespace): """ # do some sanity checking on page size before anything else if args.page_size > 1000: - console.print( - "page size must be within [0,1000]. To fetch all pages, specify --pages -1 with any legal page size" - ) + console.print("page size must be within [0,1000]. To fetch all pages, specify --pages -1 with any legal page size") sys.exit(1) s = InventorySearch(args.api_key) @@ -565,9 +537,7 @@ def cli_execute_saved_query_by_name(args: argparse.Namespace): query = results[0]["query"] try: - res = s.search( - None, query, args.page_size, None, args.sort, args.fields, args.pages - ) + res = s.search(None, query, args.page_size, None, args.sort, args.fields, args.pages) console.print_json(json.dumps(res)) except CensysAsmException: console.print("Failed to execute saved query.") @@ -582,9 +552,7 @@ def cli_execute_saved_query_by_id(args: argparse.Namespace): """ # do some sanity checking on page size before anything else if args.page_size > 1000: - console.print( - "page size must be within [0,1000]. To fetch all pages, specify --pages -1 with any legal page size" - ) + console.print("page size must be within [0,1000]. To fetch all pages, specify --pages -1 with any legal page size") sys.exit(1) s = InventorySearch(args.api_key) @@ -596,9 +564,7 @@ def cli_execute_saved_query_by_id(args: argparse.Namespace): console.print("No saved query found with that ID.") sys.exit(1) try: - res = s.search( - None, query, args.page_size, None, args.sort, args.fields, args.pages - ) + res = s.search(None, query, args.page_size, None, args.sort, args.fields, args.pages) console.print_json(json.dumps(res)) except CensysAsmException: console.print("Failed to execute saved query.") @@ -636,9 +602,7 @@ def include(parent_parser: argparse._SubParsersAction, parents: dict): parent_parser (argparse._SubParsersAction): Parent parser. parents (dict): Parent arg parsers. """ - asm_parser = parent_parser.add_parser( - "asm", description="Interact with the Censys ASM API", help="interact with ASM" - ) + asm_parser = parent_parser.add_parser("asm", description="Interact with the Censys ASM API", help="interact with ASM") def add_verbose(parser): parser.add_argument( @@ -755,9 +719,7 @@ def add_verbose(parser): type=str, default="", ) - list_parser.add_argument( - "--csv", help="output in CSV format (otherwise JSON)", action="store_true" - ) + list_parser.add_argument("--csv", help="output in CSV format (otherwise JSON)", action="store_true") add_verbose(list_parser) list_parser.set_defaults(func=cli_list_seeds) @@ -791,9 +753,7 @@ def add_verbose(parser): type=int, default=1, ) - list_saved_queries_parser.add_argument( - "--csv", help="output in CSV format (otherwise JSON)", action="store_true" - ) + list_saved_queries_parser.add_argument("--csv", help="output in CSV format (otherwise JSON)", action="store_true") add_verbose(list_saved_queries_parser) list_saved_queries_parser.set_defaults(func=cli_list_saved_queries) @@ -914,9 +874,7 @@ def add_verbose(parser): default=1, ) add_verbose(execute_saved_query_by_name_parser) - execute_saved_query_by_name_parser.set_defaults( - func=cli_execute_saved_query_by_name - ) + execute_saved_query_by_name_parser.set_defaults(func=cli_execute_saved_query_by_name) execute_saved_query_by_id_parser = asm_subparser.add_parser( "execute-saved-query-by-id", diff --git a/censys/cli/commands/config.py b/censys/cli/commands/config.py index 8f299221..684eed13 100644 --- a/censys/cli/commands/config.py +++ b/censys/cli/commands/config.py @@ -52,9 +52,7 @@ def cli_config(_: argparse.Namespace): # pragma: no cover api_id = api_id.strip() api_secret = api_secret.strip() - color = Confirm.ask( - "Do you want color output?", default=True, show_default=False, console=console - ) + color = Confirm.ask("Do you want color output?", default=True, show_default=False, console=console) config.set(DEFAULT, "color", "auto" if color else "") try: diff --git a/censys/cli/commands/hnri.py b/censys/cli/commands/hnri.py index 06be13bc..3104eb24 100644 --- a/censys/cli/commands/hnri.py +++ b/censys/cli/commands/hnri.py @@ -3,7 +3,7 @@ import argparse import sys import webbrowser -from typing import Any, List, Optional, Tuple +from typing import Any, Optional import requests from rich import box @@ -17,8 +17,8 @@ class CensysHNRI: """Searches the Censys API for the user's current IP to scan for risks.""" - HIGH_RISK_DEFINITION: List[str] = ["TELNET", "REDIS", "POSTGRES", "VNC"] - MEDIUM_RISK_DEFINITION: List[str] = ["SSH", "HTTP", "HTTPS"] + HIGH_RISK_DEFINITION: list[str] = ["TELNET", "REDIS", "POSTGRES", "VNC"] + MEDIUM_RISK_DEFINITION: list[str] = ["SSH", "HTTP", "HTTPS"] def __init__(self, api_id: Optional[str] = None, api_secret: Optional[str] = None): """Inits CensysHNRI. @@ -40,7 +40,7 @@ def get_current_ip() -> str: current_ip = str(response.json().get("ip")) return current_ip - def translate_risk(self, services: List[dict]) -> Tuple[List[dict], List[dict]]: + def translate_risk(self, services: list[dict]) -> tuple[list[dict], list[dict]]: """Interpret protocols to risks. Args: @@ -63,7 +63,7 @@ def translate_risk(self, services: List[dict]) -> Tuple[List[dict], List[dict]]: return high_risk, medium_risk - def make_risks_into_table(self, title: str, risks: List[dict]) -> Table: + def make_risks_into_table(self, title: str, risks: list[dict]) -> Table: """Creates a table of risks. Args: @@ -78,7 +78,7 @@ def make_risks_into_table(self, title: str, risks: List[dict]) -> Table: table.add_row(str(risk.get("port")), risk.get("service_name")) return table - def risks_to_string(self, high_risks: list, medium_risks: list) -> List[Any]: + def risks_to_string(self, high_risks: list, medium_risks: list) -> list[Any]: """Risks to printable string. Args: @@ -97,7 +97,7 @@ def risks_to_string(self, high_risks: list, medium_risks: list) -> List[Any]: if len_high_risk + len_medium_risk == 0: raise CensysCLIException - response: List[Any] = [] + response: list[Any] = [] if len_high_risk > 0: response.append( self.make_risks_into_table( @@ -133,9 +133,7 @@ def view_current_ip_risks(self): f"\nFor more information, please visit: https://search.censys.io/hosts/{current_ip}" # noqa: E231 ) except (CensysNotFoundException, CensysCLIException): - console.print( - "[green]:white_check_mark: No Risks were found on your network[/green]" - ) + console.print("[green]:white_check_mark: No Risks were found on your network[/green]") def cli_hnri(args: argparse.Namespace): diff --git a/censys/cli/commands/search.py b/censys/cli/commands/search.py index db74f3d6..0f224c6b 100644 --- a/censys/cli/commands/search.py +++ b/censys/cli/commands/search.py @@ -5,7 +5,7 @@ import sys import webbrowser from pathlib import Path -from typing import Any, Dict, List +from typing import Any from urllib.parse import urlencode from censys.cli.utils import V2_INDEXES, err_console, write_file @@ -13,16 +13,14 @@ from censys.search import SearchClient from censys.search.v2.api import CensysSearchAPIv2 -Results = List[dict] +Results = list[dict] DATA_DIR = Path(__file__).parent.parent / "data" HOSTS_AUTOCOMPLETE = DATA_DIR / "hosts_autocomplete.json" CERTIFICATES_AUTOCOMPLETE = DATA_DIR / "certificates_autocomplete.json" -def fields_completer( - prefix: str, parsed_args: argparse.Namespace, **kwargs -) -> List[str]: +def fields_completer(prefix: str, parsed_args: argparse.Namespace, **kwargs) -> list[str]: """Fields completer. Args: @@ -45,11 +43,7 @@ def fields_completer( return [] autocomplete_data = autocomplete_json.get("data", []) - fields = [ - field_value - for field in autocomplete_data - if not (field_value := field["value"]).endswith(".type") - ] + fields = [field_value for field in autocomplete_data if not (field_value := field["value"]).endswith(".type")] if not prefix: # Returns first 20 fields if no prefix is provided @@ -92,7 +86,7 @@ def cli_search(args: argparse.Namespace): search_args = {} write_args = {"file_format": args.format, "file_path": args.output} - results: List[Dict[str, Any]] = [] + results: list[dict[str, Any]] = [] index: CensysSearchAPIv2 = getattr(c.v2, index_type) @@ -114,9 +108,7 @@ def cli_search(args: argparse.Namespace): search_args.update({"sort": args.sort}) if args.output and not args.output.endswith(".json"): - raise CensysCLIException( - "JSON is the only valid file format for Search 2.0 responses." - ) + raise CensysCLIException("JSON is the only valid file format for Search 2.0 responses.") write_args.update( { "file_format": "json" if args.output else "screen", diff --git a/censys/cli/commands/subdomains.py b/censys/cli/commands/subdomains.py index 751cb881..a7ff273d 100644 --- a/censys/cli/commands/subdomains.py +++ b/censys/cli/commands/subdomains.py @@ -3,7 +3,6 @@ import argparse import json import sys -from typing import List, Set from censys.cli.utils import console, err_console from censys.common.exceptions import ( @@ -14,7 +13,7 @@ from censys.search import CensysCerts -def print_subdomains(subdomains: Set[str], as_json: bool = False): +def print_subdomains(subdomains: set[str], as_json: bool = False): """Print subdomains. Args: @@ -40,30 +39,20 @@ def cli_subdomains(args: argparse.Namespace): # pragma: no cover certificate_query = f"names: {args.domain}" with err_console.status(f"Querying {args.domain} subdomains"): - query = client.search( - certificate_query, per_page=100, pages=args.pages - ) # 100 is the max per page + query = client.search(certificate_query, per_page=100, pages=args.pages) # 100 is the max per page # Flatten the result, and remove duplicates for hits in query: for cert in hits: - new_subdomains: List[str] = cert.get("names", []) - subdomains.update( - [ - subdomain - for subdomain in new_subdomains - if subdomain.endswith(args.domain) - ] - ) + new_subdomains: list[str] = cert.get("names", []) + subdomains.update([subdomain for subdomain in new_subdomains if subdomain.endswith(args.domain)]) # Don't make console prints if we're in json mode if not args.json: if len(subdomains) == 0: err_console.print(f"No subdomains found for {args.domain}") return - console.print( - f"Found {len(subdomains)} unique subdomain(s) of {args.domain}" - ) + console.print(f"Found {len(subdomains)} unique subdomain(s) of {args.domain}") print_subdomains(subdomains, args.json) except CensysRateLimitExceededException: err_console.print("Censys API rate limit exceeded") @@ -95,10 +84,6 @@ def include(parent_parser: argparse._SubParsersAction, parents: dict): parents=[parents["auth"]], ) subdomains_parser.add_argument("domain", help="The base domain to search for") - subdomains_parser.add_argument( - "--pages", type=int, default=1, help="Max records to query" - ) - subdomains_parser.add_argument( - "-j", "--json", action="store_true", help="Output in JSON format" - ) + subdomains_parser.add_argument("--pages", type=int, default=1, help="Max records to query") + subdomains_parser.add_argument("-j", "--json", action="store_true", help="Output in JSON format") subdomains_parser.set_defaults(func=cli_subdomains) diff --git a/censys/cli/commands/view.py b/censys/cli/commands/view.py index 29c24f7e..8c2d5040 100644 --- a/censys/cli/commands/view.py +++ b/censys/cli/commands/view.py @@ -52,9 +52,7 @@ def cli_view(args: argparse.Namespace): ipaddress.ip_address(ip_address) except ValueError: if len(args.document_id) == 64: - err_console.print( - "This is a SHA-256 certificate fingerprint. Switching to certificates index." - ) + err_console.print("This is a SHA-256 certificate fingerprint. Switching to certificates index.") index_type = "certificates" else: raise CensysCLIException( @@ -73,9 +71,7 @@ def cli_view(args: argparse.Namespace): if index_type == "hosts": view_args["at_time"] = args.at_time else: - err_console.print( - "The --at-time option is only supported for the hosts index. Ignoring." - ) + err_console.print("The --at-time option is only supported for the hosts index. Ignoring.") document = index.view(args.document_id, **view_args) diff --git a/censys/cli/utils.py b/censys/cli/utils.py index 5f69c78f..3e154ea9 100644 --- a/censys/cli/utils.py +++ b/censys/cli/utils.py @@ -5,13 +5,13 @@ import json import os.path import sys -from typing import Any, Dict, List, Optional, Union +from typing import Any, Optional, Union from rich.console import Console from censys.common.config import DEFAULT, get_config -Results = Union[List[dict], Dict[str, Any]] +Results = Union[list[dict], dict[str, Any]] V2_INDEXES = ["hosts", "certificates"] @@ -63,7 +63,7 @@ def write_file( results_list: Results, file_format: Optional[str] = None, file_path: Optional[str] = None, - csv_fields: Optional[List[str]] = None, + csv_fields: Optional[list[str]] = None, ): """Maps formats and writes results. diff --git a/censys/common/base.py b/censys/common/base.py index 5abfa64c..41057ee4 100644 --- a/censys/common/base.py +++ b/censys/common/base.py @@ -4,7 +4,7 @@ import os import warnings from functools import wraps -from typing import Any, Callable, Optional, Type +from typing import Any, Callable, Optional import backoff import requests @@ -110,9 +110,7 @@ def __init__( "User-Agent": " ".join( [ requests.utils.default_user_agent(), - user_agent - or kwargs.get("user_agent_identifier") - or self.DEFAULT_USER_AGENT, + user_agent or kwargs.get("user_agent_identifier") or self.DEFAULT_USER_AGENT, ] ), } @@ -142,7 +140,7 @@ def request_id(self, value: Optional[str]): self._session.headers["x-request-id"] = value @staticmethod - def _get_exception_class(_: Response) -> Type[CensysAPIException]: + def _get_exception_class(_: Response) -> type[CensysAPIException]: """Maps HTTP status code or ASM error code to exception. Must be implemented by child class. @@ -157,13 +155,10 @@ def _get_exception_class(_: Response) -> Type[CensysAPIException]: @backoff.on_predicate( backoff.runtime, - predicate=lambda r: r.status_code in (408, 429, 502, 503) - and r.headers.get("Retry-After"), + predicate=lambda r: r.status_code in (408, 429, 502, 503) and r.headers.get("Retry-After"), value=lambda r: int(r.headers.get("Retry-After", 0)), ) - def _call_method( - self, method: Callable[..., Response], url: str, request_kwargs: dict - ) -> Response: + def _call_method(self, method: Callable[..., Response], url: str, request_kwargs: dict) -> Response: """Make API call. Wrapper functions for all our REST API calls checking for errors @@ -207,10 +202,7 @@ def _make_call( Returns: dict: Results from an API request. """ - if endpoint.startswith("/"): - url = f"{self._api_url}{endpoint}" - else: - url = f"{self._api_url}/{endpoint}" + url = f"{self._api_url}{endpoint}" if endpoint.startswith("/") else f"{self._api_url}/{endpoint}" request_kwargs = { "params": args or {}, @@ -240,9 +232,7 @@ def _make_call( json_data = res.json() message = json_data.get("error") or json_data.get("message") const = json_data.get("error_type") or json_data.get("status") or res.reason - error_code = json_data.get("errorCode") or json_data.get( - "statusCode", "unknown" - ) + error_code = json_data.get("errorCode") or json_data.get("statusCode", "unknown") details = json_data.get("details", "unknown") except (ValueError, json.decoder.JSONDecodeError) as error: raise CensysJSONDecodeException( diff --git a/censys/common/exceptions.py b/censys/common/exceptions.py index d9082cd3..3f779109 100644 --- a/censys/common/exceptions.py +++ b/censys/common/exceptions.py @@ -1,6 +1,6 @@ """Exceptions for Censys.""" -from typing import Dict, Optional, Type +from typing import Optional class CensysException(Exception): @@ -65,10 +65,7 @@ def __repr__(self) -> str: Returns: str: Printable representation. """ - return ( - f"{self.status_code} (Error Code: {self.error_code}), " - f"{self.message}. {self.details}" - ) + return f"{self.status_code} (Error Code: {self.error_code}), {self.message}. {self.details}" __str__ = __repr__ @@ -300,7 +297,7 @@ class CensysTooSoonToResendInviteException(CensysAsmException): class CensysExceptionMapper: """Map status code to Exception for the ASM and Search API.""" - ASM_EXCEPTIONS: Dict[int, Type[CensysAsmException]] = { + ASM_EXCEPTIONS: dict[int, type[CensysAsmException]] = { 10000: CensysMissingApiKeyException, 10001: CensysInvalidAPIKeyException, 10002: CensysInvalidAuthTokenException, @@ -355,7 +352,7 @@ class CensysExceptionMapper: } """Map of status code to ASM Exception.""" - SEARCH_EXCEPTIONS: Dict[int, Type[CensysSearchException]] = { + SEARCH_EXCEPTIONS: dict[int, type[CensysSearchException]] = { 401: CensysUnauthorizedException, 403: CensysUnauthorizedException, 404: CensysNotFoundException, diff --git a/censys/search/v1/api.py b/censys/search/v1/api.py index 6bf7295d..ae2718e0 100644 --- a/censys/search/v1/api.py +++ b/censys/search/v1/api.py @@ -2,7 +2,7 @@ import os import warnings -from typing import List, Optional, Type +from typing import Optional from requests.models import Response @@ -14,7 +14,7 @@ CensysSearchException, ) -Fields = Optional[List[str]] +Fields = Optional[list[str]] class CensysSearchAPIv1(CensysAPIBase): @@ -25,9 +25,7 @@ class CensysSearchAPIv1(CensysAPIBase): INDEX_NAME: Optional[str] = None """Name of Censys Index.""" - def __init__( - self, api_id: Optional[str] = None, api_secret: Optional[str] = None, **kwargs - ): + def __init__(self, api_id: Optional[str] = None, api_secret: Optional[str] = None, **kwargs): """Inits CensysSearchAPIv1. See CensysAPIBase for additional arguments. @@ -56,14 +54,8 @@ def __init__( config = get_config() # Try to get credentials - self._api_id = ( - api_id or os.getenv("CENSYS_API_ID") or config.get(DEFAULT, "api_id") - ) - self._api_secret = ( - api_secret - or os.getenv("CENSYS_API_SECRET") - or config.get(DEFAULT, "api_secret") - ) + self._api_id = api_id or os.getenv("CENSYS_API_ID") or config.get(DEFAULT, "api_id") + self._api_secret = api_secret or os.getenv("CENSYS_API_SECRET") or config.get(DEFAULT, "api_secret") if not self._api_id or not self._api_secret: raise CensysException("No API ID or API secret configured.") @@ -79,10 +71,8 @@ def __init__( def _get_exception_class( # type: ignore self, res: Response - ) -> Type[CensysSearchException]: - return CensysExceptionMapper.SEARCH_EXCEPTIONS.get( - res.status_code, CensysSearchException - ) + ) -> type[CensysSearchException]: + return CensysExceptionMapper.SEARCH_EXCEPTIONS.get(res.status_code, CensysSearchException) def account(self) -> dict: """Gets the current account information. diff --git a/censys/search/v2/api.py b/censys/search/v2/api.py index 7ffef32b..c8605298 100644 --- a/censys/search/v2/api.py +++ b/censys/search/v2/api.py @@ -2,8 +2,9 @@ import os import warnings +from collections.abc import Iterable, Iterator from concurrent.futures import ThreadPoolExecutor, as_completed -from typing import Any, Dict, Iterable, Iterator, List, Optional, Type, Union +from typing import Any, Optional, Union from requests.models import Response @@ -30,9 +31,7 @@ class CensysSearchAPIv2(CensysAPIBase): INDEX_NAME: str = "" """Name of Censys Index.""" - def __init__( - self, api_id: Optional[str] = None, api_secret: Optional[str] = None, **kwargs - ): + def __init__(self, api_id: Optional[str] = None, api_secret: Optional[str] = None, **kwargs): """Inits CensysSearchAPIv2. See CensysAPIBase for additional arguments. @@ -61,14 +60,8 @@ def __init__( config = get_config() # Try to get credentials - self._api_id = ( - api_id or os.getenv("CENSYS_API_ID") or config.get(DEFAULT, "api_id") - ) - self._api_secret = ( - api_secret - or os.getenv("CENSYS_API_SECRET") - or config.get(DEFAULT, "api_secret") - ) + self._api_id = api_id or os.getenv("CENSYS_API_ID") or config.get(DEFAULT, "api_id") + self._api_secret = api_secret or os.getenv("CENSYS_API_SECRET") or config.get(DEFAULT, "api_secret") if not self._api_id or not self._api_secret: raise CensysException("No API ID or API secret configured.") @@ -83,10 +76,8 @@ def __init__( def _get_exception_class( # type: ignore self, res: Response - ) -> Type[CensysSearchException]: - return CensysExceptionMapper.SEARCH_EXCEPTIONS.get( - res.status_code, CensysSearchException - ) + ) -> type[CensysSearchException]: + return CensysExceptionMapper.SEARCH_EXCEPTIONS.get(res.status_code, CensysSearchException) def account(self) -> dict: """Gets the current account's query quota. @@ -121,8 +112,8 @@ def __init__( per_page: Optional[int] = None, cursor: Optional[str] = None, pages: int = 1, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs: Any, ): """Inits Query. @@ -151,7 +142,7 @@ def __init__( self.sort = sort self.extra_args = kwargs - def __call__(self, per_page: Optional[int] = None) -> List[dict]: + def __call__(self, per_page: Optional[int] = None) -> list[dict]: """Search current index. Args: @@ -182,7 +173,7 @@ def __call__(self, per_page: Optional[int] = None) -> List[dict]: self.pages = 0 return result["hits"] - def __next__(self) -> List[dict]: + def __next__(self) -> list[dict]: """Gets next page of search results. Returns: @@ -190,7 +181,7 @@ def __next__(self) -> List[dict]: """ return self.__call__() - def __iter__(self) -> Iterator[List[dict]]: + def __iter__(self) -> Iterator[list[dict]]: """Gets Iterator. Returns: @@ -198,7 +189,7 @@ def __iter__(self) -> Iterator[List[dict]]: """ return self - def view_all(self, max_workers: int = 20) -> Dict[str, dict]: + def view_all(self, max_workers: int = 20) -> dict[str, dict]: """View each document returned from query. Please note that each result returned by the query will be looked up using the view method. @@ -237,8 +228,8 @@ def search( per_page: int = 100, cursor: Optional[str] = None, pages: int = 1, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs: Any, ) -> Query: """Search current index. @@ -265,8 +256,8 @@ def search_post_raw( query: str, per_page: int = 100, cursor: Optional[str] = None, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs, ) -> dict: """Searches the given index for all records that match the given query. @@ -300,8 +291,8 @@ def search_post( query: str, per_page: int = 100, cursor: Optional[str] = None, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs, ) -> dict: """Searches the Certs index using the POST method. @@ -334,8 +325,8 @@ def search_get_raw( query: str, per_page: int = 100, cursor: Optional[str] = None, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs: Any, ) -> dict: """Search current index using GET method. @@ -366,8 +357,8 @@ def search_get( query: str, per_page: int = 100, cursor: Optional[str] = None, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs: Any, ) -> dict: """Search current index using GET method. @@ -397,8 +388,8 @@ def raw_search( query: str, per_page: int = 100, cursor: Optional[str] = None, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs: Any, ) -> dict: """Search current index. @@ -443,10 +434,10 @@ def view(self, document_id: str, **kwargs: Any) -> dict: def bulk_view( self, - document_ids: List[str], + document_ids: list[str], max_workers: int = 20, **kwargs: Any, - ) -> Dict[str, dict]: + ) -> dict[str, dict]: """Bulk view documents from current index. View the current structured data we have on a list of documents. @@ -462,10 +453,7 @@ def bulk_view( """ documents = {} with ThreadPoolExecutor(max_workers) as executor: - threads = { - executor.submit(self.view, document_id, **kwargs): document_id - for document_id in document_ids - } + threads = {executor.submit(self.view, document_id, **kwargs): document_id for document_id in document_ids} for task in as_completed(threads): document_id = threads[task] @@ -476,9 +464,7 @@ def bulk_view( return documents - def aggregate( - self, query: str, field: str, num_buckets: int = 50, **kwargs: Any - ) -> dict: + def aggregate(self, query: str, field: str, num_buckets: int = 50, **kwargs: Any) -> dict: """Aggregate current index. Creates a report on the breakdown of the values of a field in a result set. @@ -498,7 +484,7 @@ def aggregate( # Comments - def get_comments(self, document_id: str) -> List[dict]: + def get_comments(self, document_id: str) -> list[dict]: """Get comments for a document. Args: @@ -507,9 +493,7 @@ def get_comments(self, document_id: str) -> List[dict]: Returns: List[dict]: The list of comments. """ - return self._get(self.view_path + document_id + "/comments")["result"][ - "comments" - ] + return self._get(self.view_path + document_id + "/comments")["result"]["comments"] def get_comment(self, document_id: str, comment_id: str) -> dict: """Get comment for a document. @@ -521,9 +505,7 @@ def get_comment(self, document_id: str, comment_id: str) -> dict: Returns: dict: The result set returned. """ - return self._get(self.view_path + document_id + "/comments/" + comment_id)[ - "result" - ] + return self._get(self.view_path + document_id + "/comments/" + comment_id)["result"] def add_comment(self, document_id: str, contents: str) -> dict: """Add comment to a document. @@ -535,9 +517,7 @@ def add_comment(self, document_id: str, contents: str) -> dict: Returns: dict: The result set returned. """ - return self._post( - self.view_path + document_id + "/comments", data={"contents": contents} - )["result"] + return self._post(self.view_path + document_id + "/comments", data={"contents": contents})["result"] def delete_comment(self, document_id: str, comment_id: str) -> dict: """Delete comment from a document. @@ -569,7 +549,7 @@ def update_comment(self, document_id: str, comment_id: str, contents: str) -> di # Tags - def list_all_tags(self) -> List[dict]: + def list_all_tags(self) -> list[dict]: """List all tags. Returns: @@ -587,7 +567,7 @@ def create_tag(self, name: str, color: Optional[str] = None) -> dict: Returns: dict: The result set returned. """ - tag_def: Dict[str, Any] = {"name": name} + tag_def: dict[str, Any] = {"name": name} if color: tag_def["metadata"] = {"color": color} return self._post(self.tags_path, data=tag_def)["result"] @@ -614,7 +594,7 @@ def update_tag(self, tag_id: str, name: str, color: Optional[str] = None) -> dic Returns: dict: The result set returned. """ - tag_def: Dict[str, Any] = {"name": name} + tag_def: dict[str, Any] = {"name": name} if color: tag_def["metadata"] = {"color": color} return self._put( @@ -630,9 +610,7 @@ def delete_tag(self, tag_id: str): """ self._delete(self.tags_path + "/" + tag_id) - def _list_documents_with_tag( - self, tag_id: str, endpoint: str, keyword: str - ) -> List[dict]: + def _list_documents_with_tag(self, tag_id: str, endpoint: str, keyword: str) -> list[dict]: """List documents by tag. Args: @@ -643,11 +621,9 @@ def _list_documents_with_tag( Returns: List[dict]: The list of documents. """ - return self._get(self.tags_path + "/" + tag_id + "/" + endpoint)["result"][ - keyword - ] + return self._get(self.tags_path + "/" + tag_id + "/" + endpoint)["result"][keyword] - def list_tags_on_document(self, document_id: str) -> List[dict]: + def list_tags_on_document(self, document_id: str) -> list[dict]: """List tags on a document. Args: diff --git a/censys/search/v2/certs.py b/censys/search/v2/certs.py index 1b6eceab..aaf72a82 100644 --- a/censys/search/v2/certs.py +++ b/censys/search/v2/certs.py @@ -1,7 +1,7 @@ """Interact with the Censys Search Cert API.""" import warnings -from typing import List, Optional, Union +from typing import Optional, Union from ...common.types import Datetime from ...common.utils import format_rfc3339 @@ -40,9 +40,7 @@ class CensysCerts(CensysSearchAPIv2): INDEX_NAME = "certificates" """Name of Censys Index.""" - def __init__( - self, api_id: Optional[str] = None, api_secret: Optional[str] = None, **kwargs - ): + def __init__(self, api_id: Optional[str] = None, api_secret: Optional[str] = None, **kwargs): """Inits CensysCerts. See CensysSearchAPIv2 for additional arguments. @@ -67,7 +65,7 @@ def view(self, document_id: str, **kwargs) -> dict: """ return self._get(self.view_path + document_id, args=kwargs)["result"] - def bulk_post(self, fingerprints: List[str]) -> List[dict]: + def bulk_post(self, fingerprints: list[str]) -> list[dict]: """Fetches the certificate records for the specified SHA-256 fingerprints. Using the POST method allows for a larger number of fingerprints to be queried at once. @@ -81,7 +79,7 @@ def bulk_post(self, fingerprints: List[str]) -> List[dict]: data = {"fingerprints": fingerprints} return self._post(self.bulk_path, data=data)["result"] - def bulk_get(self, fingerprints: List[str]) -> List[dict]: + def bulk_get(self, fingerprints: list[str]) -> list[dict]: """Fetches the certificate records for the specified SHA-256 fingerprints. Using the GET method allows for a smaller number of fingerprints to be queried at once. @@ -95,7 +93,7 @@ def bulk_get(self, fingerprints: List[str]) -> List[dict]: args = {"fingerprints": fingerprints} return self._get(self.bulk_path, args=args)["result"] - def bulk(self, fingerprints: List[str]) -> List[dict]: + def bulk(self, fingerprints: list[str]) -> list[dict]: """Fetches the certificate records for the specified SHA-256 fingerprints. By default, this function uses the POST method, which allows for a larger number of fingerprints to be queried at once. @@ -109,7 +107,7 @@ def bulk(self, fingerprints: List[str]) -> List[dict]: """ return self.bulk_post(fingerprints) - def bulk_view(self, fingerprints: List[str]) -> List[dict]: # type: ignore[override] + def bulk_view(self, fingerprints: list[str]) -> list[dict]: # type: ignore[override] """Fetches the certificate records for the specified SHA-256 fingerprints. By default, this function uses the POST method, which allows for a larger number of fingerprints to be queried at once. @@ -128,8 +126,8 @@ def search_post_raw( query: str, per_page: int = 50, cursor: Optional[str] = None, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs, ) -> dict: """Searches the Certs index using the POST method. Returns the raw response. @@ -159,8 +157,8 @@ def search_post( query: str, per_page: int = 50, cursor: Optional[str] = None, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs, ) -> dict: """Searches the Certs index using the POST method. @@ -193,8 +191,8 @@ def search_get( query: str, per_page: int = 50, cursor: Optional[str] = None, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs, ) -> dict: """Searches the Certs index using the GET method. @@ -224,8 +222,8 @@ def raw_search( query: str, per_page: int = 50, cursor: Optional[str] = None, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs, ) -> dict: """Searches the Certs index. @@ -259,8 +257,8 @@ def search( # type: ignore[override] per_page: int = 50, cursor: Optional[str] = None, pages: int = 1, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, **kwargs, ) -> CensysSearchAPIv2.Query: """Searches the Certs index. @@ -282,9 +280,7 @@ def search( # type: ignore[override] """ return super().search(query, per_page, cursor, pages, fields, sort, **kwargs) - def aggregate( - self, query: str, field: str, num_buckets: int = 50, **kwargs - ) -> dict: + def aggregate(self, query: str, field: str, num_buckets: int = 50, **kwargs) -> dict: """Aggregates certificate records matching a specified query into buckets based on the given field. Args: @@ -318,7 +314,7 @@ def get_hosts_by_cert(self, fingerprint: str, cursor: Optional[str] = None) -> d args = {"cursor": cursor} return self._get(self.view_path + fingerprint + "/hosts", args)["result"] - def list_certs_with_tag(self, tag_id: str) -> List[dict]: + def list_certs_with_tag(self, tag_id: str) -> list[dict]: """Returns a list of certs which are tagged with the specified tag. Args: diff --git a/censys/search/v2/hosts.py b/censys/search/v2/hosts.py index 3d1d08e3..d32aaa91 100644 --- a/censys/search/v2/hosts.py +++ b/censys/search/v2/hosts.py @@ -1,11 +1,12 @@ """Interact with the Censys Search Host API.""" -from typing import Any, Dict, List, Optional, Union +from typing import Any, Optional, Union -from .api import CensysSearchAPIv2 from censys.common.types import Datetime from censys.common.utils import format_rfc3339 +from .api import CensysSearchAPIv2 + class CensysHosts(CensysSearchAPIv2): """Interacts with the Hosts index. @@ -73,9 +74,7 @@ class CensysHosts(CensysSearchAPIv2): INDEX_NAME = "hosts" """Name of Censys Index.""" - def __init__( - self, api_id: Optional[str] = None, api_secret: Optional[str] = None, **kwargs - ): + def __init__(self, api_id: Optional[str] = None, api_secret: Optional[str] = None, **kwargs): """Inits CensysHosts. See CensysSearchAPIv2 for additional arguments. @@ -116,11 +115,11 @@ def view( def bulk_view( self, - document_ids: List[str], + document_ids: list[str], max_workers: int = 20, at_time: Optional[Datetime] = None, **kwargs: Any, - ) -> Dict[str, dict]: + ) -> dict[str, dict]: """Bulk view documents from current index. View the current structured data we have on a list of documents. @@ -145,8 +144,8 @@ def search( per_page: int = 100, cursor: Optional[str] = None, pages: int = 1, - fields: Optional[List[str]] = None, - sort: Optional[Union[str, List[str]]] = None, + fields: Optional[list[str]] = None, + sort: Optional[Union[str, list[str]]] = None, virtual_hosts: Optional[str] = None, **kwargs: Any, ) -> CensysSearchAPIv2.Query: @@ -207,9 +206,7 @@ def metadata(self) -> dict: """ return self._get(self.metadata_path)["result"] - def view_host_names( - self, ip: str, per_page: Optional[int] = None, cursor: Optional[str] = None - ) -> List[str]: + def view_host_names(self, ip: str, per_page: Optional[int] = None, cursor: Optional[str] = None) -> list[str]: """Fetches a list of host names for the specified IP address. Args: @@ -243,7 +240,7 @@ def view_host_diff( Returns: dict: A diff of the hosts. """ - args: Dict[str, Any] = {} + args: dict[str, Any] = {} if ip_b: args["ip_b"] = ip_b if at_time: @@ -284,9 +281,7 @@ def view_host_events( if end_time: args["end_time"] = format_rfc3339(end_time) - return self._get(f"/v2/experimental/{self.INDEX_NAME}/{ip}/events", args)[ - "result" - ] + return self._get(f"/v2/experimental/{self.INDEX_NAME}/{ip}/events", args)["result"] def view_host_certificates( self, @@ -312,7 +307,7 @@ def view_host_certificates( args["start_time"] = format_rfc3339(start_time) return self._get(f"/v2/{self.INDEX_NAME}/{ip}/certificates", args)["result"] - def list_hosts_with_tag(self, tag_id: str) -> List[str]: + def list_hosts_with_tag(self, tag_id: str) -> list[str]: """Returns a list of hosts which are tagged with the specified tag. Args: diff --git a/examples/search/view_host_certificates.py b/examples/search/view_host_certificates.py index 33fb0d45..8bb4cb2a 100644 --- a/examples/search/view_host_certificates.py +++ b/examples/search/view_host_certificates.py @@ -11,9 +11,7 @@ # You can also pass in a date or datetime objects. from datetime import date -certificates = h.view_host_certificates( - "1.1.1.1", per_page=1, start_time=date(2023, 1, 1) -) +certificates = h.view_host_certificates("1.1.1.1", per_page=1, start_time=date(2023, 1, 1)) print(certificates) # { # "ip": "1.1.1.1", diff --git a/examples/search/view_host_events.py b/examples/search/view_host_events.py index dc0e5166..258a08cd 100644 --- a/examples/search/view_host_events.py +++ b/examples/search/view_host_events.py @@ -11,9 +11,7 @@ # You can also pass in a date or datetime objects. from datetime import date -events = h.view_host_events( - "1.1.1.1", per_page=1, start_time=date(2022, 1, 1), end_time=date(2022, 1, 31) -) +events = h.view_host_events("1.1.1.1", per_page=1, start_time=date(2022, 1, 1), end_time=date(2022, 1, 31)) print(events) # { # 'ip': '1.1.1.1', diff --git a/poetry.lock b/poetry.lock index 822d0449..bb436488 100644 --- a/poetry.lock +++ b/poetry.lock @@ -15,18 +15,6 @@ files = [ [package.extras] test = ["coverage", "mypy", "pexpect", "ruff", "wheel"] -[[package]] -name = "astor" -version = "0.8.1" -description = "Read/rewrite/write Python ASTs" -optional = false -python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,>=2.7" -groups = ["dev"] -files = [ - {file = "astor-0.8.1-py2.py3-none-any.whl", hash = "sha256:070a54e890cefb5b3739d19f30f5a5ec840ffc9c50ffa7d23cc9fc1a38ebbfc5"}, - {file = "astor-0.8.1.tar.gz", hash = "sha256:6a6effda93f4e1ce9f618779b2dd1d9d84f1e32812c23a29b3fff6fd7f63fa5e"}, -] - [[package]] name = "backoff" version = "2.2.1" @@ -39,68 +27,6 @@ files = [ {file = "backoff-2.2.1.tar.gz", hash = "sha256:03f829f5bb1923180821643f8753b0502c3b682293992485b0eef2807afa5cba"}, ] -[[package]] -name = "black" -version = "24.8.0" -description = "The uncompromising code formatter." -optional = false -python-versions = ">=3.8" -groups = ["dev"] -files = [ - {file = "black-24.8.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:09cdeb74d494ec023ded657f7092ba518e8cf78fa8386155e4a03fdcc44679e6"}, - {file = "black-24.8.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:81c6742da39f33b08e791da38410f32e27d632260e599df7245cccee2064afeb"}, - {file = "black-24.8.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:707a1ca89221bc8a1a64fb5e15ef39cd755633daa672a9db7498d1c19de66a42"}, - {file = "black-24.8.0-cp310-cp310-win_amd64.whl", hash = "sha256:d6417535d99c37cee4091a2f24eb2b6d5ec42b144d50f1f2e436d9fe1916fe1a"}, - {file = "black-24.8.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:fb6e2c0b86bbd43dee042e48059c9ad7830abd5c94b0bc518c0eeec57c3eddc1"}, - {file = "black-24.8.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:837fd281f1908d0076844bc2b801ad2d369c78c45cf800cad7b61686051041af"}, - {file = "black-24.8.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:62e8730977f0b77998029da7971fa896ceefa2c4c4933fcd593fa599ecbf97a4"}, - {file = "black-24.8.0-cp311-cp311-win_amd64.whl", hash = "sha256:72901b4913cbac8972ad911dc4098d5753704d1f3c56e44ae8dce99eecb0e3af"}, - {file = "black-24.8.0-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:7c046c1d1eeb7aea9335da62472481d3bbf3fd986e093cffd35f4385c94ae368"}, - {file = "black-24.8.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:649f6d84ccbae73ab767e206772cc2d7a393a001070a4c814a546afd0d423aed"}, - {file = "black-24.8.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:2b59b250fdba5f9a9cd9d0ece6e6d993d91ce877d121d161e4698af3eb9c1018"}, - {file = "black-24.8.0-cp312-cp312-win_amd64.whl", hash = "sha256:6e55d30d44bed36593c3163b9bc63bf58b3b30e4611e4d88a0c3c239930ed5b2"}, - {file = "black-24.8.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:505289f17ceda596658ae81b61ebbe2d9b25aa78067035184ed0a9d855d18afd"}, - {file = "black-24.8.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:b19c9ad992c7883ad84c9b22aaa73562a16b819c1d8db7a1a1a49fb7ec13c7d2"}, - {file = "black-24.8.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1f13f7f386f86f8121d76599114bb8c17b69d962137fc70efe56137727c7047e"}, - {file = "black-24.8.0-cp38-cp38-win_amd64.whl", hash = "sha256:f490dbd59680d809ca31efdae20e634f3fae27fba3ce0ba3208333b713bc3920"}, - {file = "black-24.8.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:eab4dd44ce80dea27dc69db40dab62d4ca96112f87996bca68cd75639aeb2e4c"}, - {file = "black-24.8.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:3c4285573d4897a7610054af5a890bde7c65cb466040c5f0c8b732812d7f0e5e"}, - {file = "black-24.8.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9e84e33b37be070ba135176c123ae52a51f82306def9f7d063ee302ecab2cf47"}, - {file = "black-24.8.0-cp39-cp39-win_amd64.whl", hash = "sha256:73bbf84ed136e45d451a260c6b73ed674652f90a2b3211d6a35e78054563a9bb"}, - {file = "black-24.8.0-py3-none-any.whl", hash = "sha256:972085c618ee94f402da1af548a4f218c754ea7e5dc70acb168bfaca4c2542ed"}, - {file = "black-24.8.0.tar.gz", hash = "sha256:2500945420b6784c38b9ee885af039f5e7471ef284ab03fa35ecdde4688cd83f"}, -] - -[package.dependencies] -click = ">=8.0.0" -mypy-extensions = ">=0.4.3" -packaging = ">=22.0" -pathspec = ">=0.9.0" -platformdirs = ">=2" -tomli = {version = ">=1.1.0", markers = "python_version < \"3.11\""} -typing-extensions = {version = ">=4.0.1", markers = "python_version < \"3.11\""} - -[package.extras] -colorama = ["colorama (>=0.4.3)"] -d = ["aiohttp (>=3.7.4) ; sys_platform != \"win32\" or implementation_name != \"pypy\"", "aiohttp (>=3.7.4,!=3.9.0) ; sys_platform == \"win32\" and implementation_name == \"pypy\""] -jupyter = ["ipython (>=7.8.0)", "tokenize-rt (>=3.2.0)"] -uvloop = ["uvloop (>=0.15.2)"] - -[[package]] -name = "blacken-docs" -version = "1.18.0" -description = "Run Black on Python code blocks in documentation files." -optional = false -python-versions = ">=3.8" -groups = ["dev"] -files = [ - {file = "blacken_docs-1.18.0-py3-none-any.whl", hash = "sha256:64f592246784131e9f84dad1db397f44eeddc77fdf01726bab920a3f00a3815c"}, - {file = "blacken_docs-1.18.0.tar.gz", hash = "sha256:47bed628679d008a8eb55d112df950582e68d0f57615223929e366348d935444"}, -] - -[package.dependencies] -black = ">=22.1" - [[package]] name = "certifi" version = "2024.12.14" @@ -215,21 +141,6 @@ files = [ {file = "charset_normalizer-3.4.1.tar.gz", hash = "sha256:44251f18cd68a75b56585dd00dae26183e102cd5e0f9f1466e6df5da2ed64ea3"}, ] -[[package]] -name = "click" -version = "8.1.8" -description = "Composable command line interface toolkit" -optional = false -python-versions = ">=3.7" -groups = ["dev"] -files = [ - {file = "click-8.1.8-py3-none-any.whl", hash = "sha256:63c132bbbed01578a06712a2d1f497bb62d9c1c0d329b7903a866228027263b2"}, - {file = "click-8.1.8.tar.gz", hash = "sha256:ed53c9d8990d83c2a27deae68e4ee337473f6330c040a31d4225c9574d16096a"}, -] - -[package.dependencies] -colorama = {version = "*", markers = "platform_system == \"Windows\""} - [[package]] name = "colorama" version = "0.4.6" @@ -237,7 +148,7 @@ description = "Cross-platform colored terminal text." optional = false python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7" groups = ["dev"] -markers = "platform_system == \"Windows\" or sys_platform == \"win32\"" +markers = "sys_platform == \"win32\"" files = [ {file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"}, {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"}, @@ -331,18 +242,6 @@ tomli = {version = "*", optional = true, markers = "python_full_version <= \"3.1 [package.extras] toml = ["tomli ; python_full_version <= \"3.11.0a6\""] -[[package]] -name = "darglint" -version = "1.8.1" -description = "A utility for ensuring Google-style docstrings stay up to date with the source code." -optional = false -python-versions = ">=3.6,<4.0" -groups = ["dev"] -files = [ - {file = "darglint-1.8.1-py3-none-any.whl", hash = "sha256:5ae11c259c17b0701618a20c3da343a3eb98b3bc4b5a83d31cdd94f5ebdced8d"}, - {file = "darglint-1.8.1.tar.gz", hash = "sha256:080d5106df149b199822e7ee7deb9c012b49891538f14a11be681044f0bb20da"}, -] - [[package]] name = "exceptiongroup" version = "1.2.2" @@ -359,136 +258,6 @@ files = [ [package.extras] test = ["pytest (>=6)"] -[[package]] -name = "flake8" -version = "5.0.4" -description = "the modular source code checker: pep8 pyflakes and co" -optional = false -python-versions = ">=3.6.1" -groups = ["dev"] -files = [ - {file = "flake8-5.0.4-py2.py3-none-any.whl", hash = "sha256:7a1cf6b73744f5806ab95e526f6f0d8c01c66d7bbe349562d22dfca20610b248"}, - {file = "flake8-5.0.4.tar.gz", hash = "sha256:6fbe320aad8d6b95cec8b8e47bc933004678dc63095be98528b7bdd2a9f510db"}, -] - -[package.dependencies] -mccabe = ">=0.7.0,<0.8.0" -pycodestyle = ">=2.9.0,<2.10.0" -pyflakes = ">=2.5.0,<2.6.0" - -[[package]] -name = "flake8-black" -version = "0.3.6" -description = "flake8 plugin to call black as a code style validator" -optional = false -python-versions = ">=3.7" -groups = ["dev"] -files = [ - {file = "flake8-black-0.3.6.tar.gz", hash = "sha256:0dfbca3274777792a5bcb2af887a4cad72c72d0e86c94e08e3a3de151bb41c34"}, - {file = "flake8_black-0.3.6-py3-none-any.whl", hash = "sha256:fe8ea2eca98d8a504f22040d9117347f6b367458366952862ac3586e7d4eeaca"}, -] - -[package.dependencies] -black = ">=22.1.0" -flake8 = ">=3" -tomli = {version = "*", markers = "python_version < \"3.11\""} - -[package.extras] -develop = ["build", "twine"] - -[[package]] -name = "flake8-comprehensions" -version = "3.15.0" -description = "A flake8 plugin to help you write better list/set/dict comprehensions." -optional = false -python-versions = ">=3.8" -groups = ["dev"] -files = [ - {file = "flake8_comprehensions-3.15.0-py3-none-any.whl", hash = "sha256:b7e027bbb52be2ceb779ee12484cdeef52b0ad3c1fcb8846292bdb86d3034681"}, - {file = "flake8_comprehensions-3.15.0.tar.gz", hash = "sha256:923c22603e0310376a6b55b03efebdc09753c69f2d977755cba8bb73458a5d4d"}, -] - -[package.dependencies] -flake8 = ">=3,<3.2 || >3.2" - -[[package]] -name = "flake8-docstrings" -version = "1.7.0" -description = "Extension for flake8 which uses pydocstyle to check docstrings" -optional = false -python-versions = ">=3.7" -groups = ["dev"] -files = [ - {file = "flake8_docstrings-1.7.0-py2.py3-none-any.whl", hash = "sha256:51f2344026da083fc084166a9353f5082b01f72901df422f74b4d953ae88ac75"}, - {file = "flake8_docstrings-1.7.0.tar.gz", hash = "sha256:4c8cc748dc16e6869728699e5d0d685da9a10b0ea718e090b1ba088e67a941af"}, -] - -[package.dependencies] -flake8 = ">=3" -pydocstyle = ">=2.1" - -[[package]] -name = "flake8-isort" -version = "6.1.1" -description = "flake8 plugin that integrates isort" -optional = false -python-versions = ">=3.8" -groups = ["dev"] -files = [ - {file = "flake8_isort-6.1.1-py3-none-any.whl", hash = "sha256:0fec4dc3a15aefbdbe4012e51d5531a2eb5fa8b981cdfbc882296a59b54ede12"}, - {file = "flake8_isort-6.1.1.tar.gz", hash = "sha256:c1f82f3cf06a80c13e1d09bfae460e9666255d5c780b859f19f8318d420370b3"}, -] - -[package.dependencies] -flake8 = "*" -isort = ">=5.0.0,<6" - -[package.extras] -test = ["pytest"] - -[[package]] -name = "flake8-plugin-utils" -version = "1.3.3" -description = "The package provides base classes and utils for flake8 plugin writing" -optional = false -python-versions = ">=3.6,<4.0" -groups = ["dev"] -files = [ - {file = "flake8-plugin-utils-1.3.3.tar.gz", hash = "sha256:39f6f338d038b301c6fd344b06f2e81e382b68fa03c0560dff0d9b1791a11a2c"}, - {file = "flake8_plugin_utils-1.3.3-py3-none-any.whl", hash = "sha256:e4848c57d9d50f19100c2d75fa794b72df068666a9041b4b0409be923356a3ed"}, -] - -[[package]] -name = "flake8-pytest-style" -version = "1.7.2" -description = "A flake8 plugin checking common style issues or inconsistencies with pytest-based tests." -optional = false -python-versions = ">=3.7.2,<4.0.0" -groups = ["dev"] -files = [ - {file = "flake8_pytest_style-1.7.2-py3-none-any.whl", hash = "sha256:f5d2aa3219163a052dd92226589d45fab8ea027a3269922f0c4029f548ea5cd1"}, - {file = "flake8_pytest_style-1.7.2.tar.gz", hash = "sha256:b924197c99b951315949920b0e5547f34900b1844348432e67a44ab191582109"}, -] - -[package.dependencies] -flake8-plugin-utils = ">=1.3.2,<2.0.0" - -[[package]] -name = "flake8-simplify" -version = "0.21.0" -description = "flake8 plugin which checks for code that can be simplified" -optional = false -python-versions = ">=3.6.1" -groups = ["dev"] -files = [ - {file = "flake8_simplify-0.21.0-py3-none-any.whl", hash = "sha256:439391e762a9370b371208add0b5c5c40c3d25a98e1f5421d263215d08194183"}, - {file = "flake8_simplify-0.21.0.tar.gz", hash = "sha256:c95ff1dcc1de5949af47e0087cbf1164445881131b15bcd7a71252670f492f4d"}, -] - -[package.dependencies] -astor = ">=0.1" -flake8 = ">=3.7" - [[package]] name = "idna" version = "3.10" @@ -516,21 +285,6 @@ files = [ {file = "iniconfig-2.0.0.tar.gz", hash = "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3"}, ] -[[package]] -name = "isort" -version = "5.13.2" -description = "A Python utility / library to sort Python imports." -optional = false -python-versions = ">=3.8.0" -groups = ["dev"] -files = [ - {file = "isort-5.13.2-py3-none-any.whl", hash = "sha256:8ca5e72a8d85860d5a3fa69b8745237f2939afe12dbf656afbcb47fe72d947a6"}, - {file = "isort-5.13.2.tar.gz", hash = "sha256:48fdfcb9face5d58a4f6dde2e72a1fb8dcaf8ab26f95ab49fab84c2ddefb0109"}, -] - -[package.extras] -colors = ["colorama (>=0.4.6)"] - [[package]] name = "markdown-it-py" version = "3.0.0" @@ -556,18 +310,6 @@ profiling = ["gprof2dot"] rtd = ["jupyter_sphinx", "mdit-py-plugins", "myst-parser", "pyyaml", "sphinx", "sphinx-copybutton", "sphinx-design", "sphinx_book_theme"] testing = ["coverage", "pytest", "pytest-cov", "pytest-regressions"] -[[package]] -name = "mccabe" -version = "0.7.0" -description = "McCabe checker, plugin for flake8" -optional = false -python-versions = ">=3.6" -groups = ["dev"] -files = [ - {file = "mccabe-0.7.0-py2.py3-none-any.whl", hash = "sha256:6c2d30ab6be0e4a46919781807b4f0d834ebdd6c6e3dca0bda5a15f863427b6e"}, - {file = "mccabe-0.7.0.tar.gz", hash = "sha256:348e0240c33b60bbdf4e523192ef919f28cb2c3d7d5c7794f74009290f236325"}, -] - [[package]] name = "mdurl" version = "0.1.2" @@ -679,50 +421,6 @@ files = [ [package.extras] dev = ["jinja2"] -[[package]] -name = "pathspec" -version = "0.12.1" -description = "Utility library for gitignore style pattern matching of file paths." -optional = false -python-versions = ">=3.8" -groups = ["dev"] -files = [ - {file = "pathspec-0.12.1-py3-none-any.whl", hash = "sha256:a0d503e138a4c123b27490a4f7beda6a01c6f288df0e4a8b79c7eb0dc7b4cc08"}, - {file = "pathspec-0.12.1.tar.gz", hash = "sha256:a482d51503a1ab33b1c67a6c3813a26953dbdc71c31dacaef9a838c4e29f5712"}, -] - -[[package]] -name = "pep8-naming" -version = "0.14.1" -description = "Check PEP-8 naming conventions, plugin for flake8" -optional = false -python-versions = ">=3.8" -groups = ["dev"] -files = [ - {file = "pep8-naming-0.14.1.tar.gz", hash = "sha256:1ef228ae80875557eb6c1549deafed4dabbf3261cfcafa12f773fe0db9be8a36"}, - {file = "pep8_naming-0.14.1-py3-none-any.whl", hash = "sha256:63f514fc777d715f935faf185dedd679ab99526a7f2f503abb61587877f7b1c5"}, -] - -[package.dependencies] -flake8 = ">=5.0.0" - -[[package]] -name = "platformdirs" -version = "4.3.6" -description = "A small Python package for determining appropriate platform-specific dirs, e.g. a `user data dir`." -optional = false -python-versions = ">=3.8" -groups = ["dev"] -files = [ - {file = "platformdirs-4.3.6-py3-none-any.whl", hash = "sha256:73e575e1408ab8103900836b97580d5307456908a03e92031bab39e4554cc3fb"}, - {file = "platformdirs-4.3.6.tar.gz", hash = "sha256:357fb2acbc885b0419afd3ce3ed34564c13c9b95c89360cd9563f73aa5e2b907"}, -] - -[package.extras] -docs = ["furo (>=2024.8.6)", "proselint (>=0.14)", "sphinx (>=8.0.2)", "sphinx-autodoc-typehints (>=2.4)"] -test = ["appdirs (==1.4.4)", "covdefaults (>=2.3)", "pytest (>=8.3.2)", "pytest-cov (>=5)", "pytest-mock (>=3.14)"] -type = ["mypy (>=1.11.2)"] - [[package]] name = "pluggy" version = "1.5.0" @@ -739,48 +437,6 @@ files = [ dev = ["pre-commit", "tox"] testing = ["pytest", "pytest-benchmark"] -[[package]] -name = "pycodestyle" -version = "2.9.1" -description = "Python style guide checker" -optional = false -python-versions = ">=3.6" -groups = ["dev"] -files = [ - {file = "pycodestyle-2.9.1-py2.py3-none-any.whl", hash = "sha256:d1735fc58b418fd7c5f658d28d943854f8a849b01a5d0a1e6f3f3fdd0166804b"}, - {file = "pycodestyle-2.9.1.tar.gz", hash = "sha256:2c9607871d58c76354b697b42f5d57e1ada7d261c261efac224b664affdc5785"}, -] - -[[package]] -name = "pydocstyle" -version = "6.3.0" -description = "Python docstring style checker" -optional = false -python-versions = ">=3.6" -groups = ["dev"] -files = [ - {file = "pydocstyle-6.3.0-py3-none-any.whl", hash = "sha256:118762d452a49d6b05e194ef344a55822987a462831ade91ec5c06fd2169d019"}, - {file = "pydocstyle-6.3.0.tar.gz", hash = "sha256:7ce43f0c0ac87b07494eb9c0b462c0b73e6ff276807f204d6b53edc72b7e44e1"}, -] - -[package.dependencies] -snowballstemmer = ">=2.2.0" - -[package.extras] -toml = ["tomli (>=1.2.3) ; python_version < \"3.11\""] - -[[package]] -name = "pyflakes" -version = "2.5.0" -description = "passive checker of Python programs" -optional = false -python-versions = ">=3.6" -groups = ["dev"] -files = [ - {file = "pyflakes-2.5.0-py2.py3-none-any.whl", hash = "sha256:4579f67d887f804e67edb544428f264b7b24f435b263c4614f384135cea553d2"}, - {file = "pyflakes-2.5.0.tar.gz", hash = "sha256:491feb020dca48ccc562a8c0cbe8df07ee13078df59813b83959cbdada312ea3"}, -] - [[package]] name = "pygments" version = "2.19.1" @@ -856,21 +512,6 @@ pytest = ">=6.2.5" [package.extras] dev = ["pre-commit", "pytest-asyncio", "tox"] -[[package]] -name = "pyupgrade" -version = "3.8.0" -description = "A tool to automatically upgrade syntax for newer versions." -optional = false -python-versions = ">=3.8" -groups = ["dev"] -files = [ - {file = "pyupgrade-3.8.0-py2.py3-none-any.whl", hash = "sha256:08d0e6129f5e9da7e7a581bdbea689e0d49c3c93eeaf156a07ae2fd794f52660"}, - {file = "pyupgrade-3.8.0.tar.gz", hash = "sha256:1facb0b8407cca468dfcc1d13717e3a85aa37b9e6e7338664ad5bfe5ef50c867"}, -] - -[package.dependencies] -tokenize-rt = ">=3.2.0" - [[package]] name = "pyyaml" version = "6.0.2" @@ -997,27 +638,31 @@ typing-extensions = {version = ">=4.0.0,<5.0", markers = "python_version < \"3.1 jupyter = ["ipywidgets (>=7.5.1,<9)"] [[package]] -name = "snowballstemmer" -version = "2.2.0" -description = "This package provides 29 stemmers for 28 languages generated from Snowball algorithms." -optional = false -python-versions = "*" -groups = ["dev"] -files = [ - {file = "snowballstemmer-2.2.0-py2.py3-none-any.whl", hash = "sha256:c8e1716e83cc398ae16824e5572ae04e0d9fc2c6b985fb0f900f5f0c96ecba1a"}, - {file = "snowballstemmer-2.2.0.tar.gz", hash = "sha256:09b16deb8547d3412ad7b590689584cd0fe25ec8db3be37788be3810cbf19cb1"}, -] - -[[package]] -name = "tokenize-rt" -version = "6.0.0" -description = "A wrapper around the stdlib `tokenize` which roundtrips." +name = "ruff" +version = "0.16.5" +description = "An extremely fast Python linter and code formatter, written in Rust." optional = false -python-versions = ">=3.8" +python-versions = ">=3.7" groups = ["dev"] files = [ - {file = "tokenize_rt-6.0.0-py2.py3-none-any.whl", hash = "sha256:d4ff7ded2873512938b4f8cbb98c9b07118f01d30ac585a30d7a88353ca36d22"}, - {file = "tokenize_rt-6.0.0.tar.gz", hash = "sha256:b9711bdfc51210211137499b5e355d3de5ec88a85d2025c520cbb921b5194367"}, + {file = "ruff-0.16.5-py3-none-linux_armv6l.whl", hash = "sha256:12e5f673e774c35fbb62f288809c7653b73445f8ecec6b6063fd6ea3521aa14b"}, + {file = "ruff-0.16.5-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:eda58a5802de40e7ed5b32b64e0b32539338cc6fcd2c78f61e3ad6a0d79f51c3"}, + {file = "ruff-0.16.5-py3-none-macosx_11_0_arm64.whl", hash = "sha256:c5ae9a7b9a8875131f40f8fe967cc86abf899779efd663cb7ce3d572d01da7eb"}, + {file = "ruff-0.16.5-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:7b719b0a1f4d59710d283ab2965f621684a108a9e41da622e3b23f0326cd0025"}, + {file = "ruff-0.16.5-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:2298f2780ed1be0c5cb1361e32ab7b1467f3cce7dabe101d2210a314f2fe42e9"}, + {file = "ruff-0.16.5-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:258f29035a2dd021e7861e631b227a5b3f14e50c1184c9a6a122c5f4576154d7"}, + {file = "ruff-0.16.5-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:b9a4f0432966834019c74d1b7e5c51224305d7713f3d7faf3e7451f1a3be3cde"}, + {file = "ruff-0.16.5-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:b5eb3a8c3d0ade9cea42b591fd530368e8798380e30e0a308b85a5cf718f09ea"}, + {file = "ruff-0.16.5-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ef0f69e191a13a3c9816f63163c88790cb12cd157bbbb384e9c44745702ab105"}, + {file = "ruff-0.16.5-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:0eeab41fbea2c42f98dfb9822cdccda9d24ba38d49f6dc945b5c236d48f0ef29"}, + {file = "ruff-0.16.5-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:f0768e9df4300713fff30733c87575f68b6f1d8de41184e505b7fdd9c0c95eaf"}, + {file = "ruff-0.16.5-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:95cc70cdc7aa80c338de356279d2adbeb2de0f520b9ecd8aba75b94e95e02f91"}, + {file = "ruff-0.16.5-py3-none-musllinux_1_2_i686.whl", hash = "sha256:d185c8398ded1bfd91c0c2cb258346307571eccc473a8490af8c3977399c384a"}, + {file = "ruff-0.16.5-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:fb8e3a3c4c6a784150a7ced53b015f4b253fc2bf97a610886419ead64b4756ef"}, + {file = "ruff-0.16.5-py3-none-win32.whl", hash = "sha256:288b0a5f080492fe5635db849f9e2e84aa3cce7b7f0e955997d416c507c76a26"}, + {file = "ruff-0.16.5-py3-none-win_amd64.whl", hash = "sha256:ddc6385fb2137f616357ca03d6c74f4be987f80fed4008566b754f6032b8546f"}, + {file = "ruff-0.16.5-py3-none-win_arm64.whl", hash = "sha256:a64abe90968719b851bb7cedffaa8753fbdbdadab483089682db623f3edc587e"}, + {file = "ruff-0.16.5.tar.gz", hash = "sha256:1b88500f9ffbcab3dedb0082c9f9492e91ec3d618aac1236a3e0189938f7040b"}, ] [[package]] @@ -1112,4 +757,4 @@ zstd = ["zstandard (>=0.18.0)"] [metadata] lock-version = "2.1" python-versions = ">=3.9,<4.0" -content-hash = "8f25c415d5aa41164129c96e6b2982ec1f2f0bddfaf41c8ccea95045bc044edf" +content-hash = "a6334041f0533d60d866b13c98060d6f95896d25650902947b4c088476154b83" diff --git a/pyproject.toml b/pyproject.toml index c66901ac..f34cdf3b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -60,19 +60,7 @@ argcomplete = ">=2.0.0,<4.0.0" [tool.poetry.group.dev.dependencies] # Lint -black = ">=23.3,<25.0" -blacken-docs = "^1.13.0" -darglint = "^1.8.1" -flake8 = "^5.0.4" -flake8-black = "^0.3.6" -flake8-comprehensions = "^3.12.0" -flake8-docstrings = "^1.7.0" -flake8-isort = "^6.0.0" -flake8-pytest-style = "^1.7.2" -flake8-simplify = ">=0.20,<0.22" -isort = "^5.11.5" -pep8-naming = ">=0.13.3,<0.15.0" -pyupgrade = "^3.3.1" +ruff = ">=0.9.0" # Tests pytest = ">=7.3,<9.0" pytest-cov = ">=4,<6" @@ -83,17 +71,32 @@ parameterized = "^0.9.0" mypy = "^1.5.1" types-requests = "^2.29.0.0" -[tool.black] -target-version = ["py39"] +[tool.ruff] +line-length = 127 +target-version = "py39" -[tool.isort] -profile = "black" -line_length = 88 -multi_line_output = 3 -known_first_party = ["censys"] -known_local_folder = ["censys"] -sections = ["FUTURE", "STDLIB", "THIRDPARTY", "FIRSTPARTY", "LOCALFOLDER"] -extend_skip = ["setup.py", "conf.py"] +[tool.ruff.lint] +select = [ + "E", "W", # pycodestyle + "F", # pyflakes + "I", # isort + "N", # pep8-naming + "UP", # pyupgrade + "C4", # flake8-comprehensions + "PT", # flake8-pytest-style + "SIM", # flake8-simplify +] +ignore = [ + "E501", # line too long — ruff format is the source of truth; remaining are unfixable literals + "N818", # exception naming — renaming CensysException* would be a breaking API change + "PT028", # test default args — false positive for test utility helpers, not fixtures +] + +[tool.ruff.lint.per-file-ignores] +"examples/**" = ["E402"] + +[tool.ruff.lint.isort] +known-first-party = ["censys"] [tool.mypy] python_version = "3.9" diff --git a/tests/asm/test_api.py b/tests/asm/test_api.py index 1459e20b..1940944f 100644 --- a/tests/asm/test_api.py +++ b/tests/asm/test_api.py @@ -7,7 +7,6 @@ from pytest_mock import MockerFixture from requests.models import Response -from ..utils import CensysTestCase from censys.asm.api import CensysAsmAPI from censys.common.exceptions import ( CensysAsmException, @@ -15,6 +14,8 @@ CensysExceptionMapper, ) +from ..utils import CensysTestCase + class CensysAPIBaseTestsNoAsmEnv(unittest.TestCase): @pytest.fixture(autouse=True) @@ -31,26 +32,19 @@ def setUp(self): self.responses = responses.RequestsMock() self.responses.start() self.mocker.patch.dict("os.environ", {"CENSYS_ASM_API_KEY": ""}) - self.mock_open = self.mocker.patch( - "builtins.open", new_callable=self.mocker.mock_open, read_data="[DEFAULT]" - ) + self.mock_open = self.mocker.patch("builtins.open", new_callable=self.mocker.mock_open, read_data="[DEFAULT]") self.addCleanup(self.responses.stop) self.addCleanup(self.responses.reset) def test_no_env(self): - self.mocker.patch( - "builtins.open", new_callable=self.mock_open, read_data="[DEFAULT]" - ) + self.mocker.patch("builtins.open", new_callable=self.mock_open, read_data="[DEFAULT]") with pytest.raises(CensysException, match="No ASM API key configured."): CensysAsmAPI() class CensysAsmAPITests(CensysTestCase): - AsmExceptionParams = [ - (code, exception) - for code, exception in CensysExceptionMapper.ASM_EXCEPTIONS.items() - ] + AsmExceptionParams = [(code, exception) for code, exception in CensysExceptionMapper.ASM_EXCEPTIONS.items()] def setUp(self): super().setUp() @@ -76,8 +70,7 @@ def test_exception_repr(self): ) # Assertion assert ( - repr(exception) - == "404 (Error Code: 10014), Unable to Find Seed. [{id: 999}]" # noqa: FS003 + repr(exception) == "404 (Error Code: 10014), Unable to Find Seed. [{id: 999}]" # noqa: FS003 ) @parameterized.expand([("assets")]) diff --git a/tests/asm/test_assets.py b/tests/asm/test_assets.py index 716e6ba1..e920fea1 100644 --- a/tests/asm/test_assets.py +++ b/tests/asm/test_assets.py @@ -5,6 +5,9 @@ from parameterized import parameterized_class from pytest_mock import MockerFixture +from censys.asm.client import AsmClient +from censys.common.exceptions import CensysInvalidColorException + from .utils import ( BETA_URL, RESOURCE_PAGING_RESULTS, @@ -13,8 +16,6 @@ V1_URL, MockResponse, ) -from censys.asm.client import AsmClient -from censys.common.exceptions import CensysInvalidColorException ASSETS_URL = f"{V1_URL}/assets" BETA_ASSETS_URL = f"{BETA_URL}/assets" @@ -61,9 +62,7 @@ def __inject_fixtures(self, mocker: MockerFixture): def setUp(self): self.client = AsmClient() - self.resource_type = ( - ASSET_TYPE if self.asset_type != "subdomains" else SUBDOMAIN_ASSET_TYPE - ) + self.resource_type = ASSET_TYPE if self.asset_type != "subdomains" else SUBDOMAIN_ASSET_TYPE def get_asset_accessor(self): return getattr(self.client, self.asset_type) @@ -89,14 +88,12 @@ def test_get_assets(self): else: # Mock mock_request = self.mocker.patch("censys.common.base.requests.Session.get") - mock_request.return_value = MockResponse( - TEST_SUCCESS_CODE, self.resource_type - ) + mock_request.return_value = MockResponse(TEST_SUCCESS_CODE, self.resource_type) # Actual call assets = self.get_asset_accessor().get_assets() res = list(assets) # Assertions - assert RESOURCE_PAGING_RESULTS == res + assert res == RESOURCE_PAGING_RESULTS mock_request.assert_called_with( self.asset_type_url(), params={"pageNumber": 3, "pageSize": 500}, @@ -120,7 +117,7 @@ def test_get_assets_by_tag(self): ) res = list(assets) # Assertions - assert RESOURCE_PAGING_RESULTS == res + assert res == RESOURCE_PAGING_RESULTS mock_request.assert_called_with( self.asset_type_url(), params={ @@ -141,13 +138,9 @@ def test_get_assets_by_page(self): # Mock mock_request = self.mocker.patch("censys.common.base.requests.Session.get") - mock_request.return_value = MockResponse( - TEST_SUCCESS_CODE, self.resource_type, TEST_PAGE_NUMBER - ) + mock_request.return_value = MockResponse(TEST_SUCCESS_CODE, self.resource_type, TEST_PAGE_NUMBER) # Actual call - assets = self.get_asset_accessor().get_assets( - page_number=TEST_PAGE_NUMBER, page_size=TEST_PAGE_SIZE - ) + assets = self.get_asset_accessor().get_assets(page_number=TEST_PAGE_NUMBER, page_size=TEST_PAGE_SIZE) res = list(assets) # Assertions assert RESOURCE_PAGING_RESULTS[:6] == res @@ -179,7 +172,7 @@ def test_get_asset_comments(self): comments = self.get_asset_accessor().get_comments(self.test_asset_id) res = list(comments) # Assertions - assert RESOURCE_PAGING_RESULTS == res + assert res == RESOURCE_PAGING_RESULTS mock_request.assert_called_with( f"{self.asset_id_url()}/{COMMENT_TYPE}", params={"pageNumber": 3, "pageSize": 500}, @@ -189,13 +182,9 @@ def test_get_asset_comments(self): def test_get_asset_comments_by_page(self): # Mock mock_request = self.mocker.patch("censys.common.base.requests.Session.get") - mock_request.return_value = MockResponse( - TEST_SUCCESS_CODE, COMMENT_TYPE, TEST_PAGE_NUMBER - ) + mock_request.return_value = MockResponse(TEST_SUCCESS_CODE, COMMENT_TYPE, TEST_PAGE_NUMBER) # Actual call - comments = self.get_asset_accessor().get_comments( - self.test_asset_id, page_number=2, page_size=2 - ) + comments = self.get_asset_accessor().get_comments(self.test_asset_id, page_number=2, page_size=2) res = list(comments) # Assertions assert RESOURCE_PAGING_RESULTS[:6] == res @@ -251,9 +240,7 @@ def test_add_tag_with_color(self): mock_request = self.mocker.patch("censys.common.base.requests.Session.post") mock_request.return_value = MockResponse(TEST_SUCCESS_CODE, self.resource_type) # Actual call - self.get_asset_accessor().add_tag( - self.test_asset_id, TEST_TAG_NAME, TEST_TAG_COLOR - ) + self.get_asset_accessor().add_tag(self.test_asset_id, TEST_TAG_NAME, TEST_TAG_COLOR) # Assertions mock_request.assert_called_with( f"{self.asset_id_url()}/tags", @@ -265,9 +252,7 @@ def test_add_tag_with_color(self): def test_add_tag_with_invalid_color(self): # Actual call/error raising with pytest.raises(CensysInvalidColorException): - self.get_asset_accessor().add_tag( - self.test_asset_id, TEST_TAG_NAME, TEST_INVALID_TAG_COLOR - ) + self.get_asset_accessor().add_tag(self.test_asset_id, TEST_TAG_NAME, TEST_INVALID_TAG_COLOR) def test_add_tag_without_color(self): # Mock @@ -306,7 +291,7 @@ def test_get_subdomains(self): subdomains = self.client.domains.get_subdomains(self.test_asset_id) res = list(subdomains) # Assertions - assert RESOURCE_PAGING_RESULTS == res + assert res == RESOURCE_PAGING_RESULTS mock_request.assert_called_with( f"{self.asset_id_url()}/subdomains", params={"pageNumber": 3, "pageSize": 500}, @@ -323,7 +308,7 @@ def test_get_web_entity_instances(self): instances = self.client.web_entities.get_instances(self.test_asset_id) res = list(instances) # Assertions - assert RESOURCE_PAGING_RESULTS == res + assert res == RESOURCE_PAGING_RESULTS mock_request.assert_called_with( f"{self.asset_id_url()}/instances", params={"cursor": "test", "pageSize": None}, diff --git a/tests/asm/test_beta.py b/tests/asm/test_beta.py index 2b117d62..93cf27ca 100644 --- a/tests/asm/test_beta.py +++ b/tests/asm/test_beta.py @@ -1,8 +1,9 @@ import responses +from censys.asm import Beta + from ..utils import CensysTestCase from .utils import BETA_URL -from censys.asm import Beta TEST_LOGBOOK_DATA = { "nextWindowCursor": "string", @@ -35,9 +36,7 @@ "requestCompleteTime": "string", "totalCount": 0, "totalNewCount": 0, - "totalCountsBySubEnvironment": [ - {"environment": "string", "totalCount": 0, "totalNewCount": 0} - ], + "totalCountsBySubEnvironment": [{"environment": "string", "totalCount": 0, "totalNewCount": 0}], } TEST_HOST_COUNTS_BY_COUNTRY = { "environment": "ALL", @@ -89,9 +88,7 @@ def test_add_cloud_assets(self): json=TEST_CLOUD_ASSETS, ) # Actual call - res = self.client.add_cloud_assets( - cloud_connector_uid="uid", cloud_assets=[{"key": "value"}] - ) + res = self.client.add_cloud_assets(cloud_connector_uid="uid", cloud_assets=[{"key": "value"}]) # Assertions assert res == TEST_CLOUD_ASSETS @@ -117,9 +114,7 @@ def test_get_asset_counts(self): json=TEST_ASSET_COUNTS, ) # Actual call - res = self.client.get_asset_counts( - since="2021-01-01T00:00:00Z", environment="env", asset_type="type" - ) + res = self.client.get_asset_counts(since="2021-01-01T00:00:00Z", environment="env", asset_type="type") # Assertions assert res == TEST_ASSET_COUNTS @@ -132,9 +127,7 @@ def test_get_host_counts_by_country(self): json=TEST_HOST_COUNTS_BY_COUNTRY, ) # Actual call - res = self.client.get_host_counts_by_country( - since="2021-01-01T00:00:00Z", environment="env" - ) + res = self.client.get_host_counts_by_country(since="2021-01-01T00:00:00Z", environment="env") # Assertions assert res == TEST_HOST_COUNTS_BY_COUNTRY diff --git a/tests/asm/test_clouds.py b/tests/asm/test_clouds.py index 53362218..835e7c3e 100644 --- a/tests/asm/test_clouds.py +++ b/tests/asm/test_clouds.py @@ -1,17 +1,16 @@ import responses +from censys.asm.client import AsmClient + from ..utils import CensysTestCase from .utils import V1_URL -from censys.asm.client import AsmClient TEST_COUNT_JSON = { "totalAssetCount": 0, "totalNewAssetCount": 0, "totalCloudAssetCount": 0, "totalCloudNewAssetCount": 0, - "assetCountByProvider": [ - {"cloudProvider": "string", "assetCount": 0, "newAssetCount": 0} - ], + "assetCountByProvider": [{"cloudProvider": "string", "assetCount": 0, "newAssetCount": 0}], } diff --git a/tests/asm/test_events.py b/tests/asm/test_events.py index 0034377f..4727f1df 100644 --- a/tests/asm/test_events.py +++ b/tests/asm/test_events.py @@ -3,6 +3,9 @@ import pytest from pytest_mock import MockerFixture +from censys.asm.client import AsmClient +from censys.asm.logbook import Filters + from .utils import ( RESOURCE_PAGING_RESULTS, TEST_SUCCESS_CODE, @@ -10,8 +13,6 @@ V1_URL, MockResponse, ) -from censys.asm.client import AsmClient -from censys.asm.logbook import Filters EVENTS_URL = f"{V1_URL}/logbook" EVENTS_CURSOR_URL = f"{V1_URL}/logbook-cursor" @@ -48,9 +49,7 @@ def test_get_logbook_cursor_no_args(self): # Actual call self.client.events.get_cursor() # Assertions - mock_request.assert_called_with( - EVENTS_CURSOR_URL, params={}, timeout=TEST_TIMEOUT - ) + mock_request.assert_called_with(EVENTS_CURSOR_URL, params={}, timeout=TEST_TIMEOUT) def test_get_logbook_cursor_with_start_date(self): # Mock @@ -128,35 +127,23 @@ def test_get_logbook_cursor_with_filters_and_start_id(self): def test_get_all_events(self): # Mock mock_request = self.mocker.patch("censys.common.base.requests.Session.get") - mock_request.return_value = MockResponse( - TEST_SUCCESS_CODE, EVENTS_RESOURCE_TYPE - ) + mock_request.return_value = MockResponse(TEST_SUCCESS_CODE, EVENTS_RESOURCE_TYPE) # Actual call events = self.client.events.get_events() res = list(events) # Assertions - assert RESOURCE_PAGING_RESULTS == res - mock_request.assert_any_call( - EVENTS_URL, params={"cursor": None}, timeout=TEST_TIMEOUT - ) - mock_request.assert_any_call( - EVENTS_URL, params={"cursor": TEST_NEXT_CURSOR}, timeout=TEST_TIMEOUT - ) + assert res == RESOURCE_PAGING_RESULTS + mock_request.assert_any_call(EVENTS_URL, params={"cursor": None}, timeout=TEST_TIMEOUT) + mock_request.assert_any_call(EVENTS_URL, params={"cursor": TEST_NEXT_CURSOR}, timeout=TEST_TIMEOUT) def test_get_events_with_cursor(self): # Mock mock_request = self.mocker.patch("censys.common.base.requests.Session.get") - mock_request.return_value = MockResponse( - TEST_SUCCESS_CODE, EVENTS_RESOURCE_TYPE - ) + mock_request.return_value = MockResponse(TEST_SUCCESS_CODE, EVENTS_RESOURCE_TYPE) # Actual call events = self.client.events.get_events(TEST_CURSOR) res = list(events) # Assertions - assert RESOURCE_PAGING_RESULTS == res - mock_request.assert_any_call( - EVENTS_URL, params={"cursor": TEST_CURSOR}, timeout=TEST_TIMEOUT - ) - mock_request.assert_any_call( - EVENTS_URL, params={"cursor": TEST_NEXT_CURSOR}, timeout=TEST_TIMEOUT - ) + assert res == RESOURCE_PAGING_RESULTS + mock_request.assert_any_call(EVENTS_URL, params={"cursor": TEST_CURSOR}, timeout=TEST_TIMEOUT) + mock_request.assert_any_call(EVENTS_URL, params={"cursor": TEST_NEXT_CURSOR}, timeout=TEST_TIMEOUT) diff --git a/tests/asm/test_inventory.py b/tests/asm/test_inventory.py index ec4c1070..cc323c52 100644 --- a/tests/asm/test_inventory.py +++ b/tests/asm/test_inventory.py @@ -1,9 +1,10 @@ import responses from parameterized import parameterized +from censys.asm.inventory import InventorySearch + from ..utils import CensysTestCase from .utils import BASE_URL, WORKSPACE_ID -from censys.asm.inventory import InventorySearch INVENTORY_BASE_PATH = f"{BASE_URL}/inventory/v1" INVENTORY_SEARCH_PATH = INVENTORY_BASE_PATH @@ -29,11 +30,7 @@ "cardinality": {"field": "string"}, }, } -TEST_INVENTORY_FIELDS_JSON = { - "fields": [ - {"path": "string", "type": "string", "description": "string", "repeated": True} - ] -} +TEST_INVENTORY_FIELDS_JSON = {"fields": [{"path": "string", "type": "string", "description": "string", "repeated": True}]} class InventoryTests(CensysTestCase): diff --git a/tests/asm/test_risks.py b/tests/asm/test_risks.py index 088bd95c..f831336d 100644 --- a/tests/asm/test_risks.py +++ b/tests/asm/test_risks.py @@ -4,9 +4,10 @@ from parameterized import parameterized from responses import matchers +from censys.asm.risks import Risks + from ..utils import CensysTestCase from .utils import V2_URL -from censys.asm.risks import Risks TEST_EVENT_JSON = { "delta": "string", diff --git a/tests/asm/test_saved_queries.py b/tests/asm/test_saved_queries.py index a3f2ae74..43fad5b6 100644 --- a/tests/asm/test_saved_queries.py +++ b/tests/asm/test_saved_queries.py @@ -1,9 +1,10 @@ import responses from parameterized import parameterized +from censys.asm.saved_queries import SavedQueries + from ..utils import CensysTestCase from .utils import BASE_URL -from censys.asm.saved_queries import SavedQueries SAVED_QUERIES_BASE_PATH = f"{BASE_URL}/inventory/v1/saved-query" diff --git a/tests/asm/test_seeds.py b/tests/asm/test_seeds.py index 26f90307..f03d6c98 100644 --- a/tests/asm/test_seeds.py +++ b/tests/asm/test_seeds.py @@ -3,9 +3,10 @@ import pytest from pytest_mock import MockerFixture -from .utils import TEST_SUCCESS_CODE, TEST_TIMEOUT, V1_URL, MockResponse from censys.asm.client import AsmClient +from .utils import TEST_SUCCESS_CODE, TEST_TIMEOUT, V1_URL, MockResponse + SEEDS_URL = f"{V1_URL}/seeds" SEED_RESOURCE_TYPE = "seeds" @@ -77,9 +78,7 @@ def test_get_seeds_by_type(self): # Actual call self.client.seeds.get_seeds(seed_type=TEST_SEED_TYPE) # Assertions - mock_request.assert_called_with( - SEEDS_URL, params={"type": TEST_SEED_TYPE}, timeout=TEST_TIMEOUT - ) + mock_request.assert_called_with(SEEDS_URL, params={"type": TEST_SEED_TYPE}, timeout=TEST_TIMEOUT) def test_get_seed_by_id(self): # Mock @@ -88,9 +87,7 @@ def test_get_seed_by_id(self): # Actual call self.client.seeds.get_seed_by_id(TEST_SEED_ID) # Assertions - mock_request.assert_called_with( - f"{SEEDS_URL}/{TEST_SEED_ID}", params={}, timeout=TEST_TIMEOUT - ) + mock_request.assert_called_with(f"{SEEDS_URL}/{TEST_SEED_ID}", params={}, timeout=TEST_TIMEOUT) def test_add_seed(self): # Mock @@ -153,9 +150,7 @@ def test_replace_seeds_by_label_forced(self): mock_request = self.mocker.patch("censys.common.base.requests.Session.put") mock_request.return_value = MockResponse(TEST_SUCCESS_CODE, SEED_RESOURCE_TYPE) # Actual call - self.client.seeds.replace_seeds_by_label( - TEST_SEED_LABEL, TEST_SEED_LIST_NO_LABEL, force=True - ) + self.client.seeds.replace_seeds_by_label(TEST_SEED_LABEL, TEST_SEED_LIST_NO_LABEL, force=True) # Assertions mock_request.assert_called_with( SEEDS_URL, @@ -171,9 +166,7 @@ def test_delete_seeds_by_label(self): # Actual call self.client.seeds.delete_seeds_by_label(TEST_SEED_LABEL) # Assertions - mock_request.assert_called_with( - SEEDS_URL, params={"label": TEST_SEED_LABEL}, timeout=TEST_TIMEOUT - ) + mock_request.assert_called_with(SEEDS_URL, params={"label": TEST_SEED_LABEL}, timeout=TEST_TIMEOUT) def test_delete_seed_by_id(self): # Mock @@ -182,6 +175,4 @@ def test_delete_seed_by_id(self): # Actual call self.client.seeds.delete_seed_by_id(TEST_SEED_ID) # Assertions - mock_request.assert_called_with( - f"{SEEDS_URL}/{TEST_SEED_ID}", params={}, timeout=TEST_TIMEOUT - ) + mock_request.assert_called_with(f"{SEEDS_URL}/{TEST_SEED_ID}", params={}, timeout=TEST_TIMEOUT) diff --git a/tests/asm/utils.py b/tests/asm/utils.py index 08d90967..7787d67a 100644 --- a/tests/asm/utils.py +++ b/tests/asm/utils.py @@ -40,9 +40,7 @@ def json(self): self.json_data["endOfEvents"] = next(self.end_of_events_generator) self.json_data["pageNumber"] = next(self.number_generator) else: - self.json_data["cursor"] = ( - "test" if not next(self.end_of_events_generator) else None - ) + self.json_data["cursor"] = "test" if not next(self.end_of_events_generator) else None if "pageNumber" in self.json_data: del self.json_data["pageNumber"] self.json_data[self.resource_type] = self.get_resource() diff --git a/tests/cli/test_account.py b/tests/cli/test_account.py index 84aa511f..0d2b9d0b 100644 --- a/tests/cli/test_account.py +++ b/tests/cli/test_account.py @@ -5,11 +5,10 @@ import pytest import responses +from censys.cli import main as cli_main from tests.search.v1.test_api import ACCOUNT_JSON from tests.utils import V1_URL, CensysTestCase -from censys.cli import main as cli_main - class CensysCliAccountTest(CensysTestCase): def test_table(self): @@ -70,4 +69,4 @@ def test_json(self): cli_response = temp_stdout.getvalue().strip() # Assertions - assert ACCOUNT_JSON == json.loads(cli_response) + assert json.loads(cli_response) == ACCOUNT_JSON diff --git a/tests/cli/test_asm.py b/tests/cli/test_asm.py index 0513995a..f584cfbd 100644 --- a/tests/cli/test_asm.py +++ b/tests/cli/test_asm.py @@ -8,13 +8,12 @@ from responses import matchers from responses.matchers import json_params_matcher +from censys.cli import main as cli_main +from censys.cli.commands.asm import get_seeds_from_xml from tests.asm.utils import INVENTORY_URL, V1_URL, WORKSPACE_ID from tests.cli.test_config import TEST_CONFIG_PATH from tests.utils import CensysTestCase -from censys.cli import main as cli_main -from censys.cli.commands.asm import get_seeds_from_xml - SEEDS_JSON = [ {"value": 0, "type": "ASN"}, {"value": 0, "type": "IP_ADDRESS"}, @@ -250,9 +249,7 @@ def setUp(self): def test_add_seeds(self): # Mock - self.patch_args( - ["censys", "asm", "add-seeds", "-j", json.dumps(SEEDS_JSON)], asm_auth=True - ) + self.patch_args(["censys", "asm", "add-seeds", "-j", json.dumps(SEEDS_JSON)], asm_auth=True) self.responses.add( responses.POST, V1_URL + "/seeds", @@ -270,19 +267,13 @@ def test_add_seeds(self): def test_add_seeds_no_type(self): # Mock - self.patch_args( - ["censys", "asm", "add-seeds", "-j", json.dumps(["1.1.1.1"])], asm_auth=True - ) + self.patch_args(["censys", "asm", "add-seeds", "-j", json.dumps(["1.1.1.1"])], asm_auth=True) self.responses.add( responses.POST, V1_URL + "/seeds", status=200, json=ADD_SEEDS_JSON, - match=[ - json_params_matcher( - {"seeds": [{"value": "1.1.1.1", "type": "IP_ADDRESS", "label": ""}]} - ) - ], + match=[json_params_matcher({"seeds": [{"value": "1.1.1.1", "type": "IP_ADDRESS", "label": ""}]})], ) # Actual call @@ -428,9 +419,7 @@ def test_add_seeds_from_file_csv(self): self.mocker.patch( "builtins.open", new_callable=self.mocker.mock_open, - read_data="\n".join( - ["type,value", "IP_ADDRESS,1.1.1.1", "CIDR,192.168.0.15/24"] - ), + read_data="\n".join(["type,value", "IP_ADDRESS,1.1.1.1", "CIDR,192.168.0.15/24"]), ) self.responses.add( responses.POST, @@ -469,9 +458,7 @@ def test_add_seeds_from_file_csv_without_flag(self): self.mocker.patch( "builtins.open", new_callable=self.mocker.mock_open, - read_data="\n".join( - ["type,value", "IP_ADDRESS,1.1.1.1", "CIDR,192.168.0.15/24"] - ), + read_data="\n".join(["type,value", "IP_ADDRESS,1.1.1.1", "CIDR,192.168.0.15/24"]), ) self.responses.add( responses.POST, @@ -506,9 +493,7 @@ def test_add_seeds_invalid_json(self): def test_add_seeds_bad_json(self): # Mock - self.patch_args( - ["censys", "asm", "add-seeds", "-j", json.dumps([12345])], asm_auth=True - ) + self.patch_args(["censys", "asm", "add-seeds", "-j", json.dumps([12345])], asm_auth=True) # Actual call temp_stdout = StringIO() @@ -523,9 +508,7 @@ def test_add_seeds_bad_json(self): def test_add_seeds_partial(self): # Mock - self.patch_args( - ["censys", "asm", "add-seeds", "-j", json.dumps(SEEDS_JSON)], asm_auth=True - ) + self.patch_args(["censys", "asm", "add-seeds", "-j", json.dumps(SEEDS_JSON)], asm_auth=True) partial_json = ADD_SEEDS_JSON.copy() partial_json["addedSeeds"] = partial_json["addedSeeds"][1:] self.responses.add( @@ -546,9 +529,7 @@ def test_add_seeds_partial(self): def test_add_seeds_none(self): # Mock - self.patch_args( - ["censys", "asm", "add-seeds", "-j", json.dumps(SEEDS_JSON)], asm_auth=True - ) + self.patch_args(["censys", "asm", "add-seeds", "-j", json.dumps(SEEDS_JSON)], asm_auth=True) partial_json = ADD_SEEDS_JSON.copy() partial_json["addedSeeds"] = [] self.responses.add( @@ -567,10 +548,7 @@ def test_add_seeds_none(self): cli_main() # Assertions - assert ( - "No seeds were added. (Run with -v to get more info)" - in temp_stdout.getvalue() - ) + assert "No seeds were added. (Run with -v to get more info)" in temp_stdout.getvalue() def test_get_seeds_from_xml(self): # Actual call @@ -656,13 +634,7 @@ def test_delete_all_seeds_force_one_seed(self): responses.GET, V1_URL + "/seeds", status=200, - json={ - "seeds": [ - item - for item in GET_SEEDS_JSON["seeds"] - if item["value"] == "1.2.3.4" - ] - }, + json={"seeds": [item for item in GET_SEEDS_JSON["seeds"] if item["value"] == "1.2.3.4"]}, match=[matchers.query_param_matcher({})], ) self.responses.add( @@ -702,13 +674,7 @@ def test_delete_all_seeds_force(self): responses.GET, V1_URL + "/seeds", status=200, - json={ - "seeds": [ - item - for item in GET_SEEDS_JSON["seeds"] - if item["type"] == "IP_ADDRESS" - ] - }, + json={"seeds": [item for item in GET_SEEDS_JSON["seeds"] if item["type"] == "IP_ADDRESS"]}, match=[matchers.query_param_matcher({})], ) self.responses.add( @@ -753,13 +719,7 @@ def test_delete_all_seeds_yes(self): responses.GET, V1_URL + "/seeds", status=200, - json={ - "seeds": [ - item - for item in GET_SEEDS_JSON["seeds"] - if item["type"] == "IP_ADDRESS" - ] - }, + json={"seeds": [item for item in GET_SEEDS_JSON["seeds"] if item["type"] == "IP_ADDRESS"]}, match=[matchers.query_param_matcher({})], ) self.responses.add( @@ -774,9 +734,7 @@ def test_delete_all_seeds_yes(self): status=200, match=[matchers.query_param_matcher({})], ) - self.mocker.patch( - "builtins.input", side_effect=["y"] - ) # answer 'y' to are you sure + self.mocker.patch("builtins.input", side_effect=["y"]) # answer 'y' to are you sure # Actual call temp_stdout = StringIO() @@ -842,10 +800,7 @@ def test_delete_seeds_by_ip(self): # Assertions assert len(self.responses.calls) == 3 # make sure all three requests were seen - assert ( - "Deleted 2 seeds.\nUnable to delete 1 seeds because they were not present.\n" - in temp_stdout.getvalue() - ) + assert "Deleted 2 seeds.\nUnable to delete 1 seeds because they were not present.\n" in temp_stdout.getvalue() def test_delete_seed_by_id(self): # Mock @@ -935,9 +890,7 @@ def test_delete_seeds_by_csv(self): self.mocker.patch( "builtins.open", new_callable=self.mocker.mock_open, - read_data="\n".join( - ["type,value", "IP_ADDRESS,1.2.3.4", "CIDR,200.200.200.0/24"] - ), + read_data="\n".join(["type,value", "IP_ADDRESS,1.2.3.4", "CIDR,200.200.200.0/24"]), ) self.responses.add( @@ -987,9 +940,7 @@ def test_delete_seeds_by_asm_csv(self): self.mocker.patch( "builtins.open", new_callable=self.mocker.mock_open, - read_data="\n".join( - ["Type,Value", "IP_ADDRESS,1.2.3.4", "CIDR,200.200.200.0/24"] - ), + read_data="\n".join(["Type,Value", "IP_ADDRESS,1.2.3.4", "CIDR,200.200.200.0/24"]), ) self.responses.add( @@ -1055,10 +1006,7 @@ def test_delete_seeds_nonexistent_id(self): # Assertions assert len(self.responses.calls) == 2 # make sure all three requests were seen - assert ( - "Deleted 0 seeds.\nUnable to delete 1 seeds because they were not present.\n" - in temp_stdout.getvalue() - ) + assert "Deleted 0 seeds.\nUnable to delete 1 seeds because they were not present.\n" in temp_stdout.getvalue() def test_delete_seeds_multiple_nonexistent_id(self): # Mock @@ -1113,10 +1061,7 @@ def test_delete_seeds_multiple_nonexistent_id(self): # Assertions assert len(self.responses.calls) == 3 # make sure all three requests were seen - assert ( - "Deleted 0 seeds.\nUnable to delete 2 seeds because they were not present.\n" - in temp_stdout.getvalue() - ) + assert "Deleted 0 seeds.\nUnable to delete 2 seeds because they were not present.\n" in temp_stdout.getvalue() def test_delete_seeds_no_id_or_ip(self): # Mock @@ -1141,10 +1086,7 @@ def test_delete_seeds_no_id_or_ip(self): cli_main() # Assertions - assert ( - "Error, no seed id or value for seed.\nNo seeds to delete.\n" - in temp_stdout.getvalue() - ) + assert "Error, no seed id or value for seed.\nNo seeds to delete.\n" in temp_stdout.getvalue() def test_delete_seeds_id_and_value(self): # Mock @@ -1182,9 +1124,7 @@ def test_delete_seeds_id_and_value(self): def test_delete_labeled_seeds(self): # Mock - self.patch_args( - ["censys", "asm", "delete-labeled-seeds", "--label", "Test"], asm_auth=True - ) + self.patch_args(["censys", "asm", "delete-labeled-seeds", "--label", "Test"], asm_auth=True) self.responses.add( responses.DELETE, V1_URL + "/seeds", @@ -1268,10 +1208,7 @@ def test_replace_labeled_seeds(self): cli_main() # Assertions - assert ( - "Removed 0 seeds. Added 2 seeds. Skipped 0 reserved seeds." - in temp_stdout.getvalue() - ) + assert "Removed 0 seeds. Added 2 seeds. Skipped 0 reserved seeds." in temp_stdout.getvalue() def test_replace_labeled_seeds_without_label(self): # Mock @@ -1373,13 +1310,7 @@ def test_list_seeds_type_ip(self): responses.GET, V1_URL + "/seeds", status=200, - json={ - "seeds": [ - item - for item in GET_SEEDS_JSON["seeds"] - if item["type"] == "IP_ADDRESS" - ] - }, + json={"seeds": [item for item in GET_SEEDS_JSON["seeds"] if item["type"] == "IP_ADDRESS"]}, match=[matchers.query_param_matcher({"type": "IP_ADDRESS"})], ) @@ -1418,11 +1349,7 @@ def test_list_seeds_label_test(self): responses.GET, V1_URL + "/seeds", status=200, - json={ - "seeds": [ - item for item in GET_SEEDS_JSON["seeds"] if item["label"] == "Test" - ] - }, + json={"seeds": [item for item in GET_SEEDS_JSON["seeds"] if item["label"] == "Test"]}, match=[matchers.query_param_matcher({"label": "Test"})], ) @@ -1609,17 +1536,9 @@ def test_list_saved_queries_query_name_prefix(self): status=200, json={ "totalResults": 2, - "results": [ - query - for query in GET_SAVED_QUERIES_JSON["results"] - if "foo" in query["queryName"] - ], + "results": [query for query in GET_SAVED_QUERIES_JSON["results"] if "foo" in query["queryName"]], }, - match=[ - matchers.query_param_matcher( - {"pageSize": 50, "page": 1, "queryNamePrefix": "foo"} - ) - ], + match=[matchers.query_param_matcher({"pageSize": 50, "page": 1, "queryNamePrefix": "foo"})], ) # Actual call @@ -1657,11 +1576,7 @@ def test_list_saved_queries_filter_term(self): INVENTORY_URL + "/v1/saved-query", status=200, json=GET_SAVED_QUERIES_JSON, - match=[ - matchers.query_param_matcher( - {"pageSize": 50, "page": 1, "filterTerm": "domain"} - ) - ], + match=[matchers.query_param_matcher({"pageSize": 50, "page": 1, "filterTerm": "domain"})], ) # Actual call diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py index 7f73a5dd..686ff222 100644 --- a/tests/cli/test_cli.py +++ b/tests/cli/test_cli.py @@ -3,11 +3,10 @@ import pytest -from tests.utils import CensysTestCase - from censys.cli import main as cli_main from censys.cli.commands import __all__ as cli_commands from censys.common import __version__ +from tests.utils import CensysTestCase class CensysCliTest(CensysTestCase): diff --git a/tests/cli/test_config.py b/tests/cli/test_config.py index 3f22d00f..3bd8da27 100644 --- a/tests/cli/test_config.py +++ b/tests/cli/test_config.py @@ -1,9 +1,6 @@ import pytest import responses -from tests.search.v1.test_api import ACCOUNT_JSON -from tests.utils import V1_URL, CensysTestCase - from censys.cli import main as cli_main from censys.common.config import ( CENSYS_PATH, @@ -12,6 +9,8 @@ default_config, get_config, ) +from tests.search.v1.test_api import ACCOUNT_JSON +from tests.utils import V1_URL, CensysTestCase TEST_CONFIG_PATH = CONFIG_PATH + ".test" @@ -29,9 +28,7 @@ def prompt_side_effect(arg, **kwargs): def confirm_side_effect(arg, **kwargs): - if arg == "Do you want color output?": - return True - return False + return arg == "Do you want color output?" class CensysConfigCliTest(CensysTestCase): @@ -109,9 +106,7 @@ def test_search_config_makedirs(self): mock_makedirs.assert_called_with(CENSYS_PATH) def test_config_default(self): - mock_isfile = self.mocker.patch( - "censys.common.config.os.path.isfile", return_value=True - ) + mock_isfile = self.mocker.patch("censys.common.config.os.path.isfile", return_value=True) config = get_config() mock_isfile.return_value = False mock_isfile.assert_called_with(TEST_CONFIG_PATH) @@ -126,9 +121,7 @@ def test_search_config_custom_config(self): "config", ] ) - self.mocker.patch.dict( - "censys.common.config.os.environ", {"CENSYS_CONFIG_PATH": "censys.cfg"} - ) + self.mocker.patch.dict("censys.common.config.os.environ", {"CENSYS_CONFIG_PATH": "censys.cfg"}) self.responses.add( responses.GET, diff --git a/tests/cli/test_hnri.py b/tests/cli/test_hnri.py index cb9d0b87..ae19f354 100644 --- a/tests/cli/test_hnri.py +++ b/tests/cli/test_hnri.py @@ -4,12 +4,11 @@ import pytest import responses -from tests.search.v2.test_hosts import VIEW_HOST_JSON -from tests.utils import V2_URL, CensysTestCase - from censys.cli import main as cli_main from censys.cli.commands.hnri import CensysHNRI from censys.common.exceptions import CensysCLIException +from tests.search.v2.test_hosts import VIEW_HOST_JSON +from tests.utils import V2_URL, CensysTestCase class CensysCliHNRITest(CensysTestCase): diff --git a/tests/cli/test_search.py b/tests/cli/test_search.py index 362f69f4..edb23f7d 100644 --- a/tests/cli/test_search.py +++ b/tests/cli/test_search.py @@ -4,7 +4,7 @@ import os from io import StringIO from pathlib import Path -from typing import Dict, Optional, Tuple +from typing import Optional from urllib.parse import urlencode import pytest @@ -13,14 +13,6 @@ from requests import PreparedRequest from responses import matchers -from tests.search.v2.test_certs import SEARCH_CERTS_JSON -from tests.search.v2.test_hosts import ( - SEARCH_HOSTS_JSON, - SERVER_ERROR_JSON, - TOO_MANY_REQUESTS_ERROR_JSON, -) -from tests.utils import V2_URL, CensysTestCase - from censys.cli import main as cli_main from censys.cli.commands.search import ( CERTIFICATES_AUTOCOMPLETE, @@ -28,11 +20,18 @@ fields_completer, ) from censys.common.exceptions import CensysCLIException, CensysException +from tests.search.v2.test_certs import SEARCH_CERTS_JSON +from tests.search.v2.test_hosts import ( + SEARCH_HOSTS_JSON, + SERVER_ERROR_JSON, + TOO_MANY_REQUESTS_ERROR_JSON, +) +from tests.utils import V2_URL, CensysTestCase WROTE_PREFIX = "Wrote results to file" -def search_callback(request: PreparedRequest) -> Tuple[int, Dict[str, str], str]: +def search_callback(request: PreparedRequest) -> tuple[int, dict[str, str], str]: payload = json.loads(request.body) # type: ignore resp_body = { "result": { @@ -63,16 +62,10 @@ def test_search_help(self): def test_no_creds(self): # Mock self.patch_args(["censys", "search", "test"]) - self.mocker.patch( - "builtins.open", new_callable=self.mocker.mock_open, read_data="[DEFAULT]" - ) - self.mocker.patch.dict( - "os.environ", {"CENSYS_API_ID": "", "CENSYS_API_SECRET": ""} - ) + self.mocker.patch("builtins.open", new_callable=self.mocker.mock_open, read_data="[DEFAULT]") + self.mocker.patch.dict("os.environ", {"CENSYS_API_ID": "", "CENSYS_API_SECRET": ""}) # Actual Call/Assertion - with pytest.raises( - CensysException, match="No API ID or API secret configured." - ): + with pytest.raises(CensysException, match="No API ID or API secret configured."): cli_main() def test_invalid_timeout(self): @@ -430,9 +423,7 @@ def test_open_certificates(self): # Actual call/error raising with pytest.raises(SystemExit, match="0"): cli_main() - query_str = urlencode( - {"q": "domain: censys.io AND ports: 443", "resource": "certificates"} - ) + query_str = urlencode({"q": "domain: censys.io AND ports: 443", "resource": "certificates"}) # Assertions mock_open.assert_called_with( f"https://search.censys.io/search?{query_str}" # noqa: E231 @@ -480,15 +471,11 @@ def test_fields_completer( else: expected_fields = json.load(autocomplete_file.open())["data"] expected_fields = [ - field_value - for field in expected_fields - if not (field_value := field["value"]).endswith(".type") + field_value for field in expected_fields if not (field_value := field["value"]).endswith(".type") ] if prefix == "": expected_fields = expected_fields[:20] - assert ( - fields_completer(prefix=prefix, parsed_args=parsed_args) == expected_fields - ) + assert fields_completer(prefix=prefix, parsed_args=parsed_args) == expected_fields if __name__ == "__main__": diff --git a/tests/cli/test_subdomains.py b/tests/cli/test_subdomains.py index 1e6cf5cb..cb74fe6d 100644 --- a/tests/cli/test_subdomains.py +++ b/tests/cli/test_subdomains.py @@ -1,14 +1,12 @@ import contextlib from io import StringIO -from typing import Set import responses from parameterized import parameterized -from tests.utils import V2_URL, CensysTestCase - from censys.cli import main as cli_main from censys.cli.commands import subdomains +from tests.utils import V2_URL, CensysTestCase TEST_DOMAINS = { "help.censys.io", @@ -36,7 +34,7 @@ class CensysCliSubdomainsTest(CensysTestCase): def test_print_subdomains( self, test_json_bool: bool, - test_subdomains: Set[str] = TEST_DOMAINS, + test_subdomains: set[str] = TEST_DOMAINS, ): # Mock mock_print_json = self.mocker.patch("censys.cli.utils.console.print_json") @@ -58,11 +56,7 @@ def test_search_subdomains(self): responses.POST, V2_URL + "/certificates/search", json=CERT_SEARCH_RESPONSE, - match=[ - responses.matchers.json_params_matcher( - {"per_page": 100, "q": "names: censys.io"} - ) - ], + match=[responses.matchers.json_params_matcher({"per_page": 100, "q": "names: censys.io"})], ) self.mocker.patch( "argparse._sys.argv", diff --git a/tests/cli/test_utils.py b/tests/cli/test_utils.py index e2a289c8..d97b9721 100644 --- a/tests/cli/test_utils.py +++ b/tests/cli/test_utils.py @@ -4,9 +4,8 @@ import pytest from parameterized import parameterized -from tests.utils import CensysTestCase - from censys.cli.utils import valid_datetime_type +from tests.utils import CensysTestCase class CensysCliUtilsTest(CensysTestCase): diff --git a/tests/cli/test_view.py b/tests/cli/test_view.py index 7f3c1bcc..52129d43 100644 --- a/tests/cli/test_view.py +++ b/tests/cli/test_view.py @@ -7,14 +7,13 @@ import pytest import responses +from censys.cli import main as cli_main +from censys.common.exceptions import CensysCLIException, CensysException from tests.cli.test_search import WROTE_PREFIX from tests.search.v2.test_certs import VIEW_CERT_JSON from tests.search.v2.test_hosts import VIEW_HOST_JSON from tests.utils import V2_URL, CensysTestCase -from censys.cli import main as cli_main -from censys.common.exceptions import CensysCLIException, CensysException - class CensysCliViewTest(CensysTestCase): def test_search_help(self): @@ -30,16 +29,10 @@ def test_search_help(self): def test_no_creds(self): # Mock self.patch_args(["censys", "view", "test"]) - self.mocker.patch( - "builtins.open", new_callable=mock_open, read_data="[DEFAULT]" - ) - self.mocker.patch.dict( - "os.environ", {"CENSYS_API_ID": "", "CENSYS_API_SECRET": ""} - ) + self.mocker.patch("builtins.open", new_callable=mock_open, read_data="[DEFAULT]") + self.mocker.patch.dict("os.environ", {"CENSYS_API_ID": "", "CENSYS_API_SECRET": ""}) # Actual call - with pytest.raises( - CensysException, match="No API ID or API secret configured." - ): + with pytest.raises(CensysException, match="No API ID or API secret configured."): cli_main() def test_write_json(self): @@ -247,8 +240,7 @@ def test_incorrect_index_type_certs(self): ) self.responses.add( responses.GET, - V2_URL - + "/certificates/9b00121b4e85d50667ded1a8aa39855771bdb67ceca6f18726b49374b41f0041", + V2_URL + "/certificates/9b00121b4e85d50667ded1a8aa39855771bdb67ceca6f18726b49374b41f0041", status=200, json=VIEW_CERT_JSON, ) @@ -292,8 +284,7 @@ def test_has_at_time_for_certs(self): ) self.responses.add( responses.GET, - V2_URL - + "/certificates/9b00121b4e85d50667ded1a8aa39855771bdb67ceca6f18726b49374b41f0041", + V2_URL + "/certificates/9b00121b4e85d50667ded1a8aa39855771bdb67ceca6f18726b49374b41f0041", status=200, json=VIEW_CERT_JSON, ) diff --git a/tests/search/v1/test_api.py b/tests/search/v1/test_api.py index 9f60eed6..3288f508 100644 --- a/tests/search/v1/test_api.py +++ b/tests/search/v1/test_api.py @@ -6,14 +6,13 @@ from parameterized import parameterized from requests.models import Response -from tests.utils import V1_URL, CensysTestCase - from censys.common.exceptions import ( CensysException, CensysExceptionMapper, CensysSearchException, ) from censys.search.v1.api import CensysSearchAPIv1 +from tests.utils import V1_URL, CensysTestCase ACCOUNT_JSON = { "login": "test@censys.io", @@ -23,10 +22,7 @@ "quota": {"used": 1, "resets_at": "2021-01-01 01:00:00", "allowance": 100}, } -SearchExceptionParams = [ - (code, exception) - for code, exception in CensysExceptionMapper.SEARCH_EXCEPTIONS.items() -] +SearchExceptionParams = [(code, exception) for code, exception in CensysExceptionMapper.SEARCH_EXCEPTIONS.items()] class CensysSearchAPITests(CensysTestCase): @@ -73,7 +69,5 @@ def test_exception_repr(self): class CensysAPIBaseTestsNoSearchEnv(unittest.TestCase): @patch("builtins.open", new_callable=mock_open, read_data="[DEFAULT]") def test_no_env(self, mock_file): - with pytest.raises( - CensysException, match="No API ID or API secret configured." - ): + with pytest.raises(CensysException, match="No API ID or API secret configured."): CensysSearchAPIv1() diff --git a/tests/search/v1/test_data.py b/tests/search/v1/test_data.py index 6b75ed75..571e4902 100644 --- a/tests/search/v1/test_data.py +++ b/tests/search/v1/test_data.py @@ -1,8 +1,7 @@ import responses -from tests.utils import V1_URL, CensysTestCase - from censys.search import SearchClient +from tests.utils import V1_URL, CensysTestCase SERIES_JSON = { "primary_series": "", diff --git a/tests/search/v2/test_api.py b/tests/search/v2/test_api.py index 432c72d4..55e6955d 100644 --- a/tests/search/v2/test_api.py +++ b/tests/search/v2/test_api.py @@ -6,16 +6,12 @@ from parameterized import parameterized from requests.models import Response -from tests.search.v1.test_api import ACCOUNT_JSON -from tests.utils import V1_URL, CensysTestCase - from censys.common.exceptions import CensysException, CensysExceptionMapper from censys.search.v2.api import CensysSearchAPIv2 +from tests.search.v1.test_api import ACCOUNT_JSON +from tests.utils import V1_URL, CensysTestCase -SearchExceptionParams = [ - (code, exception) - for code, exception in CensysExceptionMapper.SEARCH_EXCEPTIONS.items() -] +SearchExceptionParams = [(code, exception) for code, exception in CensysExceptionMapper.SEARCH_EXCEPTIONS.items()] class CensysSearchAPITests(CensysTestCase): @@ -50,7 +46,5 @@ def test_account_and_quota(self): class CensysAPIBaseTestsNoSearchEnv(unittest.TestCase): @patch("builtins.open", new_callable=mock_open, read_data="[DEFAULT]") def test_no_env(self, mock_file): - with pytest.raises( - CensysException, match="No API ID or API secret configured." - ): + with pytest.raises(CensysException, match="No API ID or API secret configured."): CensysSearchAPIv2() diff --git a/tests/search/v2/test_certs.py b/tests/search/v2/test_certs.py index 932256d0..4ebf6dc3 100644 --- a/tests/search/v2/test_certs.py +++ b/tests/search/v2/test_certs.py @@ -1,13 +1,12 @@ from datetime import datetime -from typing import Any, Dict, List, Optional +from typing import Any, Optional import responses from parameterized import parameterized from responses import matchers -from tests.utils import V2_URL, CensysTestCase - from censys.search import SearchClient +from tests.utils import V2_URL, CensysTestCase TEST_CERT = "fb444eb8e68437bae06232b9f5091bccff62a768ca09e92eb5c9c2cf9d17c426" ALTERNATE_CERT = "9b00121b4e85d50667ded1a8aa39855771bdb67ceca6f18726b49374b41f0041" @@ -296,8 +295,8 @@ def test_search_post(self, method_name: str, raw: bool = False): ) def test_search( self, - fields: Optional[List[str]] = None, - sort: Optional[List[str]] = None, + fields: Optional[list[str]] = None, + sort: Optional[list[str]] = None, cursor: Optional[str] = None, ): self.responses.add( @@ -306,9 +305,7 @@ def test_search( status=200, json=SEARCH_CERTS_JSON, ) - query = self.api.search( - TEST_SEARCH_QUERY, fields=fields, sort=sort, cursor=cursor - ) + query = self.api.search(TEST_SEARCH_QUERY, fields=fields, sort=sort, cursor=cursor) assert next(query) == SEARCH_CERTS_JSON["result"]["hits"] @parameterized.expand( @@ -361,7 +358,7 @@ def test_search( ), ] ) - def test_search_get(self, params: Dict[str, Any], expected_params: Dict[str, Any]): + def test_search_get(self, params: dict[str, Any], expected_params: dict[str, Any]): self.responses.add( responses.GET, f"{V2_URL}/certificates/search", diff --git a/tests/search/v2/test_comments.py b/tests/search/v2/test_comments.py index c4e60025..ef3cb1bc 100644 --- a/tests/search/v2/test_comments.py +++ b/tests/search/v2/test_comments.py @@ -1,10 +1,9 @@ import responses from parameterized import parameterized_class -from tests.utils import V2_URL, CensysTestCase - from censys.search.v2 import CensysCerts, CensysHosts from censys.search.v2.api import CensysSearchAPIv2 +from tests.utils import V2_URL, CensysTestCase TEST_COMMENT = "**This is a comment.**" GET_COMMENTS_RESPONSE = { diff --git a/tests/search/v2/test_hosts.py b/tests/search/v2/test_hosts.py index a6961084..6cecab9a 100644 --- a/tests/search/v2/test_hosts.py +++ b/tests/search/v2/test_hosts.py @@ -1,17 +1,16 @@ import datetime import json from copy import deepcopy -from typing import Any, Dict, List, Optional +from typing import Any, Optional import pytest import responses from parameterized import parameterized from responses import matchers -from tests.utils import V2_URL, CensysTestCase - from censys.common.exceptions import CensysInternalServerException from censys.search import CensysHosts, SearchClient +from tests.utils import V2_URL, CensysTestCase TEST_HOST = "8.8.8.8" TEST_SEARCH_QUERY = "services.service_name: HTTP" @@ -161,8 +160,7 @@ }, } RATE_LIMIT_ERROR_JSON = { - "error": "Rate limit exceeded. See https://search.censys.io/account " - "for rate limit details.", + "error": "Rate limit exceeded. See https://search.censys.io/account for rate limit details.", "status": "error", "error_type": "rate_limit_exceeded", } @@ -324,7 +322,7 @@ def test_search_post(self, method_name: str, raw: bool = False): ) def test_search( self, - fields: Optional[List[str]] = None, + fields: Optional[list[str]] = None, sort: Optional[str] = None, cursor: Optional[str] = None, virtual_hosts: Optional[str] = None, @@ -351,11 +349,7 @@ def test_search_per_page(self): V2_URL + "/hosts/search", status=200, json=SEARCH_HOSTS_JSON, - match=[ - matchers.json_params_matcher( - {"q": "services.service_name: HTTP", "per_page": test_per_page} - ) - ], + match=[matchers.json_params_matcher({"q": "services.service_name: HTTP", "per_page": test_per_page})], ) query = self.api.search("services.service_name: HTTP", per_page=test_per_page) @@ -401,7 +395,7 @@ def test_search_per_page(self): ), ] ) - def test_search_get(self, params: Dict[str, Any], expected_params: Dict[str, Any]): + def test_search_get(self, params: dict[str, Any], expected_params: dict[str, Any]): self.responses.add( responses.GET, f"{V2_URL}/hosts/search", @@ -458,11 +452,7 @@ def test_search_pages(self): V2_URL + "/hosts/search", status=200, json=SEARCH_HOSTS_JSON, - match=[ - matchers.json_params_matcher( - {"q": "services.service_name: HTTP", "per_page": 100} - ) - ], + match=[matchers.json_params_matcher({"q": "services.service_name: HTTP", "per_page": 100})], ) page_2_json = deepcopy(SEARCH_HOSTS_JSON) hits = page_2_json["result"]["hits"] @@ -515,11 +505,7 @@ def request_callback(_): V2_URL + "/hosts/search", status=200, json=SEARCH_HOSTS_JSON, - match=[ - matchers.json_params_matcher( - {"q": "services.service_name: HTTP", "per_page": 100} - ) - ], + match=[matchers.json_params_matcher({"q": "services.service_name: HTTP", "per_page": 100})], ) self.responses.add_callback( responses.POST, @@ -538,11 +524,7 @@ def test_search_pages_retry_fail(self): V2_URL + "/hosts/search", status=200, json=SEARCH_HOSTS_JSON, - match=[ - matchers.json_params_matcher( - {"q": "services.service_name: HTTP", "per_page": 100} - ) - ], + match=[matchers.json_params_matcher({"q": "services.service_name: HTTP", "per_page": 100})], ) self.responses.add( responses.POST, @@ -602,23 +584,18 @@ def test_search_fields(self): ], ) - query = self.api.search( - "services.service_name: HTTP", fields=["ip", "services.port"] - ) + query = self.api.search("services.service_name: HTTP", fields=["ip", "services.port"]) assert query() == SEARCH_HOSTS_JSON["result"]["hits"] def test_aggregate(self): self.responses.add( responses.GET, - V2_URL - + "/hosts/aggregate?field=services.port&q=services.service_name: HTTP&num_buckets=4", + V2_URL + "/hosts/aggregate?field=services.port&q=services.service_name: HTTP&num_buckets=4", status=200, json=AGGREGATE_HOSTS_JSON, ) self.maxDiff = None - res = self.api.aggregate( - "services.service_name: HTTP", "services.port", num_buckets=4 - ) + res = self.api.aggregate("services.service_name: HTTP", "services.port", num_buckets=4) assert res == AGGREGATE_HOSTS_JSON["result"] diff --git a/tests/search/v2/test_tags.py b/tests/search/v2/test_tags.py index 4b2bd073..5df1f502 100644 --- a/tests/search/v2/test_tags.py +++ b/tests/search/v2/test_tags.py @@ -2,10 +2,9 @@ import responses from parameterized import parameterized_class -from tests.utils import BASE_URL, CensysTestCase - from censys.search.v2 import CensysCerts, CensysHosts from censys.search.v2.api import CensysSearchAPIv2 +from tests.utils import BASE_URL, CensysTestCase TEST_TAG_NAME = "is-honeypot" TEST_TAG_COLOR = "#ff0000" @@ -39,9 +38,7 @@ LIST_HOSTS_RESPONSE = { "code": 200, "status": "OK", - "result": { - "hosts": [{"ip": "1.1.1.1", "tagged_at": "2021-01-01T12:00:00.000000Z"}] - }, + "result": {"hosts": [{"ip": "1.1.1.1", "tagged_at": "2021-01-01T12:00:00.000000Z"}]}, } LIST_CERTS_RESPONSE = { "code": 200, @@ -93,11 +90,7 @@ def test_create_tag(self): BASE_URL + self.api.tags_path, status=200, json=CREATE_TAG_RESPONSE, - match=[ - responses.json_params_matcher( - {"name": TEST_TAG_NAME, "metadata": {"color": TEST_TAG_COLOR}} - ) - ], + match=[responses.json_params_matcher({"name": TEST_TAG_NAME, "metadata": {"color": TEST_TAG_COLOR}})], ) results = self.api.create_tag(TEST_TAG_NAME, TEST_TAG_COLOR) assert results == CREATE_TAG_RESPONSE["result"] @@ -118,11 +111,7 @@ def test_update_tag(self): BASE_URL + self.api.tags_path + "/" + TEST_TAG_ID, status=200, json=CREATE_TAG_RESPONSE, - match=[ - responses.json_params_matcher( - {"name": TEST_TAG_NAME, "metadata": {"color": TEST_TAG_COLOR}} - ) - ], + match=[responses.json_params_matcher({"name": TEST_TAG_NAME, "metadata": {"color": TEST_TAG_COLOR}})], ) results = self.api.update_tag(TEST_TAG_ID, TEST_TAG_NAME, TEST_TAG_COLOR) assert results == CREATE_TAG_RESPONSE["result"] @@ -174,9 +163,7 @@ def test_list_hosts_with_tag(self): json=LIST_HOSTS_RESPONSE, ) results = self.api.list_hosts_with_tag(TEST_TAG_ID) - assert results == [ - host["ip"] for host in LIST_HOSTS_RESPONSE["result"]["hosts"] - ] + assert results == [host["ip"] for host in LIST_HOSTS_RESPONSE["result"]["hosts"]] def test_list_certs_with_tag(self): if self.index == "hosts": diff --git a/tests/test_base.py b/tests/test_base.py index 4fbce94e..fdeebe7e 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -3,11 +3,12 @@ import responses from requests.models import Response -from .utils import CensysTestCase from censys.common import __version__ from censys.common.base import CensysAPIBase from censys.common.exceptions import CensysAPIException, CensysException +from .utils import CensysTestCase + TEST_URL = "https://url" TEST_ENDPOINT = "/endpoint" ERROR_JSON = { @@ -73,28 +74,20 @@ def test_invalid_json_response(self): # Actual call base = CensysAPIBase(TEST_URL) # Assertion/error raising - with pytest.raises( - CensysAPIException, match="is not valid JSON and cannot be decoded" - ): + with pytest.raises(CensysAPIException, match="is not valid JSON and cannot be decoded"): base._get(TEST_ENDPOINT) def test_default_user_agent(self): # Mock/actual call base = CensysAPIBase(TEST_URL) # Assertions - assert ( - base._session.headers["User-Agent"] - == f"{requests.utils.default_user_agent()} censys-python/{__version__}" - ) + assert base._session.headers["User-Agent"] == f"{requests.utils.default_user_agent()} censys-python/{__version__}" def test_user_agent(self): # Mock/actual call base = CensysAPIBase(TEST_URL, user_agent="test") # Assertions - assert ( - base._session.headers["User-Agent"] - == requests.utils.default_user_agent() + " test" - ) + assert base._session.headers["User-Agent"] == requests.utils.default_user_agent() + " test" def test_request_id(self): id_value = "my-request-id" diff --git a/tests/test_client.py b/tests/test_client.py index cfb83d2b..54062a9b 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,6 +1,7 @@ -from .utils import CensysTestCase from censys.search import SearchClient +from .utils import CensysTestCase + ALL_INDEXES = { "v1": ["data"], "v2": ["hosts", "certificates"], diff --git a/tests/utils.py b/tests/utils.py index 18a33df3..c0dc3cfa 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -1,5 +1,5 @@ import unittest -from typing import List, Optional +from typing import Optional import pytest import responses @@ -55,7 +55,7 @@ def setUpApi(self, api: CensysAPIBase): # noqa: N802 def patch_args( self, - args: List[str], + args: list[str], search_auth: Optional[bool] = False, asm_auth: Optional[bool] = False, ):