Repository navigation
TPT-4472: Create tests for stream commands - #901
psnoch-akamai wants to merge 4 commits into
Conversation
psnoch-akamai
changed the title
ef2bcee to
46eb9cb
Compare
There was a problem hiding this comment.
🟡 Changes recommended
Negative tests assume ID 1 is invalid and can delete unrelated destinations or create unmanaged streams, while fixture defects cause setup failures and incomplete cleanup.
5 open issues
What changed in this PR
Adds integration coverage for the CLI’s stream and destination commands, including error handling and resource lifecycle checks.
Changes:
- Adds help, listing, and negative-command tests, with smoke coverage for listings.
- Exercises destination creation, viewing, history, and deletion, plus an opt-in long-running stream test.
- Adds resource fixtures, status polling, and shared command registration.
| File | Description |
|---|---|
tests/integration/streams/test_streams.py |
Tests stream help, listing, errors, creation, and viewing. |
tests/integration/streams/test_destination.py |
Tests destination listing, errors, lifecycle, and history. |
tests/integration/streams/fixtures.py |
Provides storage-key, destination, and stream fixtures with cleanup and polling. |
tests/integration/helpers.py |
Registers streams in the shared command map. |
Review effort: Balanced
| host = os.getenv("LINODE_CLI_API_HOST") | ||
| if token is None: |
There was a problem hiding this comment.
I don't see any issue here
There was a problem hiding this comment.
All 3 of your IF conditionals here checks token value, instead of token, host, version. Moreover, host and version seem to be optional so there is no need to specify/check them unless you want to use specific host/version (e.g. v4beta)
| @pytest.fixture | ||
| def create_object_storage_keys(): | ||
| label = get_random_text(8) + "_destination_cli_test" | ||
| test_bucket = get_object_storage_buckets()["data"][0] |
There was a problem hiding this comment.
46eb9cb to
ef70629
Compare
| "--details.access_key_secret", | ||
| create_object_storage_keys[1], |
There was a problem hiding this comment.
…est_create_view_stream
ef70629 to
3202789
Compare
There was a problem hiding this comment.
🔵 Needs a closer look
The new integration fixtures fail during setup when optional API overrides are absent or the account has no existing Object Storage buckets, preventing the destination and stream success tests from exercising the commands.
3 open issues
Review effort: Balanced
| @pytest.fixture | ||
| def create_object_storage_keys(): | ||
| label = get_random_text(8) + "_destination_cli_test" | ||
| if get_object_storage_buckets()["data"][0] is None: |
There was a problem hiding this comment.
It will fail if there is no buckets at all (then data=[] leads to 'list index out of range' error). Moreover, wouldnt be better to create a bucket instead of checking if there is any already? It would let you run the tests no matter if it exists or not
|
|
||
| # handle `--flag=value` form | ||
| if "=" in arg and arg.split("=", 1)[0] in SENSITIVE_ARG_NAMES: | ||
| flag, _ = arg.split("=", 1) |
There was a problem hiding this comment.
you can avoid "_" using [0] similarly as it is in IF:
flag = arg.split("=", 1)[0]
| result_view_json["destinations"][0]["type"] == "akamai_object_storage" | ||
| ) | ||
| assert result_view_json["status"] == "active" | ||
| assert result_view_json["type"] == "audit_logs" |
There was a problem hiding this comment.
Can we extract these assertions into separate function? Additionally, is it enough to just check existence of access_key_id, bucket_name, host, path etc. with no specific values?
| ) | ||
|
|
||
|
|
||
| def wait_for_stream_status( |
There was a problem hiding this comment.
Is it possible to use existing wait_for_condition?
| import pytest | ||
| import requests | ||
|
|
||
| from linodecli import ExitCodes |
📝 Description
Create tests for stream commands
✔️ How to Test