Resolve linting errors from ruff

Author: jonavellecuerdo

tests/conftest.py

@@ -1,35 +1,35 @@
-import os
-
 import pytest
 import vcr
 from click.testing import CliRunner
 
 from tim.opensearch import configure_opensearch_client
 
+EXIT_CODES = {
+    "success": 0,
+    "error": 1,
+    "invalid_command": 2,
+}
 my_vcr = vcr.VCR(
     cassette_library_dir="tests/fixtures/cassettes",
     filter_headers=["authorization"],
 )
 
 
 @pytest.fixture(autouse=True)
-def test_env():
-    os.environ = {
-        "AWS_ACCESS_KEY_ID": "test",
-        "AWS_SECRET_ACCESS_KEY": "test",
-        "AWS_SESSION_TOKEN": "test",
-        "TIMDEX_OPENSEARCH_ENDPOINT": "localhost",
-        "SENTRY_DSN": None,
-        "WORKSPACE": "test",
-    }
-    yield
-
-
-@pytest.fixture()
+def _test_env(monkeypatch):
+    monkeypatch.setenv("SENTRY_DSN", "None")
+    monkeypatch.setenv("WORKSPACE", "test")
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "test")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "test")
+    monkeypatch.setenv("AWS_SESSION_TOKEN", "test")
+    monkeypatch.setenv("TIMDEX_OPENSEARCH_ENDPOINT", "localhost")
+
+
+@pytest.fixture
 def test_opensearch_client():
     return configure_opensearch_client("localhost")
 
 
-@pytest.fixture()
+@pytest.fixture
 def runner():
     return CliRunner()

tests/test_cli.py

@@ -4,7 +4,7 @@
 
 from tim.cli import main
 
-from .conftest import my_vcr
+from .conftest import EXIT_CODES, my_vcr
 
 
 def escape_ansi(line):
@@ -19,7 +19,7 @@ def test_main_group_no_options_configures_correctly_and_invokes_result_callback(
 ):
     monkeypatch.delenv("TIMDEX_OPENSEARCH_ENDPOINT", raising=False)
     result = runner.invoke(main, ["ping"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Logger 'root' configured with level=INFO" in caplog.text
     assert "OpenSearch client configured for endpoint 'localhost'" in caplog.text
     assert "Total time to complete process" in caplog.text
@@ -31,7 +31,7 @@ def test_main_group_all_options_configures_correctly_and_invokes_result_callback
 ):
     monkeypatch.delenv("TIMDEX_OPENSEARCH_ENDPOINT", raising=False)
     result = runner.invoke(main, ["--verbose", "--url", "localhost", "ping"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Logger 'root' configured with level=DEBUG" in caplog.text
     assert "OpenSearch client configured for endpoint 'localhost'" in caplog.text
     assert "Total time to complete process" in caplog.text
@@ -42,7 +42,7 @@ def test_main_group_options_from_env_configures_correctly_and_invokes_result_cal
     caplog, runner
 ):
     result = runner.invoke(main, ["ping"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Logger 'root' configured with level=INFO" in caplog.text
     assert "OpenSearch client configured for endpoint 'localhost'" in caplog.text
     assert "Total time to complete process" in caplog.text
@@ -51,27 +51,27 @@ def test_main_group_options_from_env_configures_correctly_and_invokes_result_cal
 @my_vcr.use_cassette("get_aliases.yaml")
 def test_aliases(runner):
     result = runner.invoke(main, ["aliases"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Alias: alias-with-multiple-indexes" in result.stdout
 
 
 @my_vcr.use_cassette("get_indexes.yaml")
 def test_indexes(runner):
     result = runner.invoke(main, ["indexes"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Name: index-with-multiple-aliases" in result.stdout
 
 
 @my_vcr.use_cassette("ping_localhost.yaml")
 def test_ping(runner):
     result = runner.invoke(main, ["ping"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Name: docker-cluster" in result.stdout
 
 
 def test_create_index_neither_name_nor_source_passed(runner):
     result = runner.invoke(main, ["create"])
-    assert result.exit_code == 2
+    assert result.exit_code == EXIT_CODES["invalid_command"]
     assert "Must provide either a name or source for the new index." in result.stdout
 
 
@@ -80,7 +80,7 @@ def test_create_index_name_and_source_passed(runner):
         main,
         ["create", "--index", "aspace-2022-09-01t12-34-56", "--source", "aspace"],
     )
-    assert result.exit_code == 2
+    assert result.exit_code == EXIT_CODES["invalid_command"]
     assert (
         "Only one of --index and --source options is allowed, not both."
         in escape_ansi(result.stdout)
@@ -89,18 +89,18 @@ def test_create_index_name_and_source_passed(runner):
 
 def test_create_index_invalid_name_passed(runner):
     result = runner.invoke(main, ["create", "--index", "wrong"])
-    assert result.exit_code == 2
+    assert result.exit_code == EXIT_CODES["invalid_command"]
 
 
 def test_create_index_invalid_source_passed(runner):
     result = runner.invoke(main, ["create", "--source", "wrong"])
-    assert result.exit_code == 2
+    assert result.exit_code == EXIT_CODES["invalid_command"]
 
 
 @my_vcr.use_cassette("cli/create_index_exists.yaml")
 def test_create_index_exists(caplog, runner):
     result = runner.invoke(main, ["create", "--index", "aspace-2022-09-20t15-59-38"])
-    assert result.exit_code == 1
+    assert result.exit_code == EXIT_CODES["error"]
     assert (
         "tim.cli",
         40,
@@ -113,37 +113,37 @@ def test_create_index_exists(caplog, runner):
 @my_vcr.use_cassette("cli/create_index_success.yaml")
 def test_create_index_success(caplog, runner):
     result = runner.invoke(main, ["create", "--source", "aspace"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Index 'aspace-2022-09-01t00-00-00' created." in caplog.text
 
 
 @my_vcr.use_cassette("delete_success.yaml")
 def test_delete_index_with_force(runner):
     result = runner.invoke(main, ["delete", "-i", "test-index", "-f"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Index 'test-index' deleted." in result.stdout
 
 
 @my_vcr.use_cassette("delete_success.yaml")
 def test_delete_index_with_confirmation(monkeypatch, runner):
     monkeypatch.setattr("builtins.input", lambda _: "y")
     result = runner.invoke(main, ["delete", "-i", "test-index"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Index 'test-index' deleted." in result.stdout
 
 
 @my_vcr.use_cassette("delete_without_confirmation.yaml")
 def test_delete_index_without_confirmation(monkeypatch, runner):
     monkeypatch.setattr("builtins.input", lambda _: "n")
     result = runner.invoke(main, ["delete", "-i", "test-index"])
-    assert result.exit_code == 1
+    assert result.exit_code == EXIT_CODES["error"]
     assert "Ok, index will not be deleted." in result.stdout
 
 
 @my_vcr.use_cassette("demote_no_aliases_for_index.yaml")
 def test_demote_index_no_aliases_for_index(runner):
     result = runner.invoke(main, ["demote", "-i", "test-index"])
-    assert result.exit_code == 1
+    assert result.exit_code == EXIT_CODES["error"]
     assert (
         "Index 'test-index' has no aliases, please check aliases and try again."
         in result.stdout
@@ -154,29 +154,29 @@ def test_demote_index_no_aliases_for_index(runner):
 def test_demote_index_from_primary_alias_with_confirmation(monkeypatch, runner):
     monkeypatch.setattr("builtins.input", lambda _: "y")
     result = runner.invoke(main, ["demote", "-i", "test-index"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Index 'test-index' demoted from aliases: ['all-current']" in result.stdout
 
 
 @my_vcr.use_cassette("demote_from_primary_alias_without_confirmation.yaml")
 def test_demote_index_from_primary_alias_without_confirmation(monkeypatch, runner):
     monkeypatch.setattr("builtins.input", lambda _: "n")
     result = runner.invoke(main, ["demote", "-i", "test-index"])
-    assert result.exit_code == 1
+    assert result.exit_code == EXIT_CODES["error"]
     assert "Ok, index will not be demoted." in result.stdout
 
 
 @my_vcr.use_cassette("demote_no_primary_alias.yaml")
 def test_demote_index_no_primary_alias(runner):
     result = runner.invoke(main, ["demote", "-i", "test-index"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Index 'test-index' demoted from aliases: ['not-primary']" in result.stdout
 
 
 @my_vcr.use_cassette("promote_index.yaml")
 def test_promote_index(caplog, runner):
     result = runner.invoke(main, ["promote", "-i", "testsource-index"])
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert "Index promoted" in caplog.text
 
 
@@ -195,7 +195,7 @@ def test_bulk_index_with_index_name_success(caplog, runner):
             "tests/fixtures/sample_records.json",
         ],
     )
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert (
         "Bulk indexing records from file 'tests/fixtures/sample_records.json' into "
         "index 'dspace-2022-09-01t00-00-00'" in caplog.text
@@ -210,7 +210,7 @@ def test_bulk_index_with_source_success(caplog, runner):
         main,
         ["bulk-index", "--source", "dspace", "tests/fixtures/sample_records.json"],
     )
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert (
         "Bulk indexing records from file 'tests/fixtures/sample_records.json' into "
         "index 'dspace-2022-09-01t00-00-00'" in caplog.text
@@ -230,7 +230,7 @@ def test_bulk_delete_with_index_name_success(caplog, runner):
             "tests/fixtures/sample_deleted_records.txt",
         ],
     )
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert (
         "Bulk deleting records in file 'tests/fixtures/sample_deleted_records.txt' "
         "from index 'alma-2022-09-01t00-00-00'" in caplog.text
@@ -250,7 +250,7 @@ def test_bulk_delete_with_source_success(caplog, runner):
             "tests/fixtures/sample_deleted_records.txt",
         ],
     )
-    assert result.exit_code == 0
+    assert result.exit_code == EXIT_CODES["success"]
     assert (
         "Bulk deleting records in file 'tests/fixtures/sample_deleted_records.txt' "
         "from index 'alma-2022-09-01t00-00-00'" in caplog.text

tests/test_config.py

@@ -18,14 +18,15 @@ def test_configure_index_settings():
 def test_configure_logger_not_verbose():
     logger = logging.getLogger(__name__)
     result = configure_logger(logger, verbose=False)
-    assert logger.getEffectiveLevel() == 20
+
+    assert logger.getEffectiveLevel() == logging.INFO
     assert result == "Logger 'tests.test_config' configured with level=INFO"
 
 
 def test_configure_logger_verbose():
     logger = logging.getLogger(__name__)
     result = configure_logger(logger, verbose=True)
-    assert logger.getEffectiveLevel() == 10
+    assert logger.getEffectiveLevel() == logging.DEBUG
     assert result == "Logger 'tests.test_config' configured with level=DEBUG"
 
 

tests/test_helpers.py

@@ -9,18 +9,18 @@
 
 def test_confirm_action_yes(monkeypatch):
     monkeypatch.setattr("builtins.input", lambda _: "Y")
-    assert helpers.confirm_action("test-index", "delete") is True
+    assert helpers.confirm_action("delete test-index") is True
 
 
 def test_confirm_action_no(monkeypatch):
     monkeypatch.setattr("builtins.input", lambda _: "n")
-    assert helpers.confirm_action("test-index", "delete") is False
+    assert helpers.confirm_action("delete test-index") is False
 
 
 def test_confirm_action_invalid(capsys, monkeypatch):
     inputs = iter(["wrong", "y"])
     monkeypatch.setattr("builtins.input", lambda _: next(inputs))
-    assert helpers.confirm_action("test-index", "delete") is True
+    assert helpers.confirm_action("delete test-index") is True
     out, _ = capsys.readouterr()
     assert out == "Invalid input: 'wrong', must be one of 'y' or 'n'.\n"
 
@@ -54,9 +54,8 @@ def test_generate_bulk_actions_delete():
 def test_generate_bulk_actions_invalid_action_raises_error():
     records = [{"timdex_record_id": "12345", "other_fields": "some_data"}]
     actions = helpers.generate_bulk_actions("test-index", records, "wrong")
-    with pytest.raises(ValueError) as error:
+    with pytest.raises(ValueError, match="Invalid action parameter"):
         next(actions)
-        assert "Invalid action parameter" in str(error.value)
 
 
 def test_get_source_from_index():
@@ -69,53 +68,56 @@ def test_get_source_from_index_without_dash():
 
 def test_parse_records():
     records = list(helpers.parse_records("tests/fixtures/sample_records.json"))
-    assert len(records) == 6
+    n_sample_records = 6
+    assert len(records) == n_sample_records
     assert isinstance(records[0], dict)
 
 
 def test_parse_deleted_records():
     records = list(
         helpers.parse_deleted_records("tests/fixtures/sample_deleted_records.txt")
     )
-    assert len(records) == 3
+    n_sample_deleted_records = 3
+    assert len(records) == n_sample_deleted_records
     assert isinstance(records[0], dict)
 
 
 def test_validate_bulk_cli_options_neither_index_nor_source_passed(
     test_opensearch_client,
 ):
-    with pytest.raises(UsageError) as error:
+    with pytest.raises(
+        UsageError, match="Must provide either an existing index name or a valid source."
+    ):
         helpers.validate_bulk_cli_options(None, None, test_opensearch_client)
-    assert "Must provide either an existing index name or a valid source." == str(
-        error.value
-    )
 
 
 def test_validate_bulk_cli_options_index_and_source_passed(test_opensearch_client):
-    with pytest.raises(UsageError) as error:
+    with pytest.raises(
+        UsageError, match="Only one of --index and --source options is allowed, not both."
+    ):
         helpers.validate_bulk_cli_options(
             "index-name", "source-name", test_opensearch_client
         )
-    assert "Only one of --index and --source options is allowed, not both." == str(
-        error.value
-    )
 
 
 @my_vcr.use_cassette("helpers/bulk_cli_nonexistent_index.yaml")
 def test_validate_bulk_cli_options_nonexistent_index_passed(test_opensearch_client):
-    with pytest.raises(BadParameter) as error:
+    with pytest.raises(
+        BadParameter, match="Index 'wrong' does not exist in the cluster."
+    ):
         helpers.validate_bulk_cli_options("wrong", None, test_opensearch_client)
-    assert "Index 'wrong' does not exist in the cluster." == str(error.value)
 
 
 @my_vcr.use_cassette("helpers/bulk_cli_no_primary_index_for_source.yaml")
 def test_validate_bulk_cli_options_no_primary_index_for_source(test_opensearch_client):
-    with pytest.raises(BadParameter) as error:
+    with pytest.raises(
+        BadParameter,
+        match=(
+            "No index name was passed and there is no "
+            "primary-aliased index for source 'dspace'."
+        ),
+    ):
         helpers.validate_bulk_cli_options(None, "dspace", test_opensearch_client)
-    assert (
-        "No index name was passed and there is no primary-aliased index for source "
-        "'dspace'." == str(error.value)
-    )
 
 
 def test_validate_index_name_no_value():