Skip to content

chore(xtest): Improve tool use output capture #323

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
dmihalcik-virtru wants to merge 4 commits into
base: main
Choose a base branch
from
Draft

chore(xtest): Improve tool use output capture #323

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c1afd93
Update test_policytypes.py
dmihalcik-virtru Jul 14, 2025
e31a92d
black
dmihalcik-virtru Jul 14, 2025
9d10dca
Update abac.py
dmihalcik-virtru Jul 14, 2025
342f23e
Update cli.sh
dmihalcik-virtru Jul 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 44 additions & 57 deletions xtest/abac.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,14 +262,13 @@ def __init__(self, otdfctl_path: str | None = None):
def kas_registry_list(self) -> list[KasEntry]:
cmd = self.otdfctl + "policy kas-registry list".split()
logger.info(f"kr-ls [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
if out:
print(out)
assert code == 0
assert process.returncode == 0
o = json.loads(out)
if not o:
return []
Expand All @@ -285,7 +284,7 @@ def kas_registry_create(
if public_key:
cmd += [f"--public-keys={public_key.model_dump_json()}"]
logger.info(f"kr-create [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
Expand All @@ -308,7 +307,7 @@ def kas_registry_keys_list(self, kas: KasEntry) -> list[KasKey]:
cmd = self.otdfctl + "policy kas-registry key list".split()
cmd += [f"--kas={kas.uri}"]
logger.info(f"kr-keys-ls [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
Expand Down Expand Up @@ -338,7 +337,7 @@ def kas_registry_create_public_key_only(
f"--key-id={public_key.kid}",
f"--algorithm={public_key.algStr}",
]
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
Expand All @@ -355,7 +354,7 @@ def key_assign_ns(self, key: KasKey, ns: Namespace) -> NamespaceKey:
f"--namespace={ns.id}",
]
logger.info(f"key-assign [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
Expand All @@ -372,14 +371,13 @@ def grant_assign_ns(self, kas: KasEntry, ns: Namespace) -> KasGrantNamespace:
f"--namespace-id={ns.id}",
]
logger.info(f"grant-update [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
if out:
print(out)
assert code == 0
assert process.returncode == 0
return KasGrantNamespace.model_validate_json(out)

def key_assign_attr(self, key: KasKey, attr: Attribute) -> AttributeKey:
Expand All @@ -389,7 +387,7 @@ def key_assign_attr(self, key: KasKey, attr: Attribute) -> AttributeKey:
f"--attribute={attr.id}",
]
logger.info(f"key-assign [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
Expand All @@ -406,14 +404,13 @@ def grant_assign_attr(self, kas: KasEntry, attr: Attribute) -> KasGrantAttribute
f"--attribute-id={attr.id}",
]
logger.info(f"grant-update [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
if out:
print(out)
assert code == 0
assert process.returncode == 0
return KasGrantAttribute.model_validate_json(out)

def key_assign_value(self, key: KasKey, val: AttributeValue) -> ValueKey:
Expand All @@ -423,7 +420,7 @@ def key_assign_value(self, key: KasKey, val: AttributeValue) -> ValueKey:
f"--value={val.id}",
]
logger.info(f"key-assign [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
Expand All @@ -440,14 +437,13 @@ def grant_assign_value(self, kas: KasEntry, val: AttributeValue) -> KasGrantValu
f"--value-id={val.id}",
]
logger.info(f"grant-update [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
if out:
print(out)
assert code == 0
assert process.returncode == 0
return KasGrantValue.model_validate_json(out)

def key_unassign_ns(self, key: KasKey, ns: Namespace) -> NamespaceKey:
Expand All @@ -457,14 +453,13 @@ def key_unassign_ns(self, key: KasKey, ns: Namespace) -> NamespaceKey:
f"--namespace={ns.id}",
]
logger.info(f"key-assign [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
if out:
print(out)
assert code == 0
assert process.returncode == 0
return NamespaceKey.model_validate_json(out)

# Deprecated in otdfctl 0.22
Expand All @@ -475,14 +470,13 @@ def grant_unassign_ns(self, kas: KasEntry, ns: Namespace) -> KasGrantNamespace:
f"--namespace-id={ns.id}",
]
logger.info(f"grant-update [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
if out:
print(out)
assert code == 0
assert process.returncode == 0
return KasGrantNamespace.model_validate_json(out)

def key_unassign_attr(self, key: KasKey, attr: Attribute) -> AttributeKey:
Expand All @@ -492,7 +486,7 @@ def key_unassign_attr(self, key: KasKey, attr: Attribute) -> AttributeKey:
f"--attribute={attr.id}",
]
logger.info(f"key-assign [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
Expand All @@ -509,14 +503,13 @@ def grant_unassign_attr(self, kas: KasEntry, attr: Attribute) -> KasGrantAttribu
f"--attribute-id={attr.id}",
]
logger.info(f"grant-update [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
if out:
print(out)
assert code == 0
assert process.returncode == 0
return KasGrantAttribute.model_validate_json(out)

def key_unassign_value(self, key: KasKey, val: AttributeValue) -> ValueKey:
Expand All @@ -526,7 +519,7 @@ def key_unassign_value(self, key: KasKey, val: AttributeValue) -> ValueKey:
f"--value={val.id}",
]
logger.info(f"key-assign [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
Expand All @@ -543,27 +536,25 @@ def grant_unassign_value(self, kas: KasEntry, val: AttributeValue) -> KasGrantVa
f"--value-id={val.id}",
]
logger.info(f"grant-update [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
if out:
print(out)
assert code == 0
assert process.returncode == 0
return KasGrantValue.model_validate_json(out)

def namespace_list(self) -> list[Namespace]:
cmd = self.otdfctl + "policy attributes namespaces list".split()
logger.info(f"ns-ls [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = process.communicate()
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

if err:
print(err, file=sys.stderr, flush=code != 0)
print(err, file=sys.stderr, flush=True)
if out:
print(out, flush=code != 0)
assert code == 0
print(out, flush=True)
assert process.returncode == 0
o = json.loads(out)
if not o:
return []
Expand All @@ -573,14 +564,13 @@ def namespace_create(self, name: str) -> Namespace:
cmd = self.otdfctl + "policy attributes namespaces create".split()
cmd += [f"--name={name}"]
logger.info(f"ns-create [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr, flush=code != 0)
print(err, file=sys.stderr, flush=True)
if out:
print(out, flush=code != 0)
assert code == 0
print(out, flush=True)
assert process.returncode == 0
return Namespace.model_validate_json(out)

def attribute_create(
Expand All @@ -596,14 +586,13 @@ def attribute_create(
if values:
cmd += [f"--value={','.join(values)}"]
logger.info(f"attr-create [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr, flush=code != 0)
print(err, file=sys.stderr, flush=True)
if out:
print(out, flush=code != 0)
assert code == 0
print(out, flush=True)
assert process.returncode == 0
return Attribute.model_validate_json(out)

def scs_create(self, scs: list[SubjectSet]) -> SubjectConditionSet:
Expand All @@ -612,14 +601,13 @@ def scs_create(self, scs: list[SubjectSet]) -> SubjectConditionSet:
cmd += [f"--subject-sets=[{','.join([s.model_dump_json() for s in scs])}]"]

logger.info(f"scs-create [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr, flush=code != 0)
print(err, file=sys.stderr, flush=True)
if out:
print(out, flush=code != 0)
assert code == 0
print(out, flush=True)
assert process.returncode == 0
return SubjectConditionSet.model_validate_json(out)

def scs_map(
Expand All @@ -638,20 +626,19 @@ def scs_map(
]

logger.info(f"sm-create [{' '.join(cmd)}]")
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
code = process.wait()
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider logging the error output using the logger instead of printing to stderr. This allows for consistent logging and easier debugging.[^1]

logger.error(err)

out, err = process.communicate()
if err:
print(err, file=sys.stderr)
if out:
print(out)
if (
code != 0
process.returncode != 0
and not self.flag_scs_map_action_standard
and err.find(b"--action-standard") >= 0
and err and err.find(b"--action-standard") >= 0
):
self.flag_scs_map_action_standard = True
return self.scs_map(sc, value)

assert code == 0
assert process.returncode == 0
return SubjectMapping.model_validate_json(out)
31 changes: 30 additions & 1 deletion xtest/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,43 @@
import secrets
import assertions
import json
import logging
import sys
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization
from pathlib import Path
from pydantic_core import to_jsonable_python

import abac
import tdfs
from typing import cast
from typing import cast, Any


# Configure pytest to capture all subprocess output
@pytest.hookimpl(trylast=True)
def pytest_configure(config: Any) -> None:
# Set log capture level to debug to ensure all logs are captured
if hasattr(config, "option"):
config.option.log_cli = True
config.option.log_cli_level = "DEBUG"
config.option.log_cli_format = (
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

# Configure logging to ensure we see everything from subprocesses
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stdout, # Ensure output goes to stdout for pytest capture
)
# Make sure subprocess logger is set to DEBUG level
logging.getLogger("subprocess").setLevel(logging.DEBUG)

# Set environment variable to make sure Python doesn't buffer stdout/stderr
os.environ["PYTHONUNBUFFERED"] = "1"

# Set buffering mode for subprocess to line buffered
os.environ["PYTHONFAULTHANDLER"] = "1"
Comment on lines +23 to +46
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jul 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

This pytest_configure hook is redundant with the settings in the new xtest/pytest.ini file. To avoid duplication and keep configuration in a single, declarative place, this function should be removed.

  • The log_cli* options are set in pytest.ini.
  • The env section in pytest.ini sets PYTHONUNBUFFERED and PYTHONFAULTHANDLER.
  • The logging.basicConfig call is also made redundant by pytest's logging setup when using log_cli.



def englist(s: tuple[str]) -> str:
Expand Down
19 changes: 19 additions & 0 deletions xtest/pytest.ini
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
[pytest]
# Display all logs during test runs
log_cli = True
log_cli_level = DEBUG
log_cli_format = %(asctime)s - %(name)s - %(levelname)s - %(message)s

# Configure subprocess settings for better output capture
pythonpath = .
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# Log all test activity including setup and teardown
log_level = DEBUG

# Set buffer size to 0 for immediate output
env =
PYTHONUNBUFFERED=1
PYTHONFAULTHANDLER=1
2 changes: 1 addition & 1 deletion xtest/sdk/js/cli.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ if [ "$1" == "supports" ]; then
nano_attribute_bug)
set -o pipefail
# Versions released unable to set nanotdf attributes
npx $CTL --version | jq -re '.["@opentdf/sdk"]' | awk -F. '{ if ($1 > 0 || ($1 == 0 && $2 > 4) || ($1 == 0 && $2 == 4 && $3 > 0)) exit 1; else exit 0; }'
npx $CTL --version | jq -re '.["@opentdf/sdk"]' | awk -F. '{ if ($1 > 0 || ($1 == 0 && $2 > 4) || ($1 == 0 && $2 == 4 && $3 >= 0)) exit 1; else exit 0; }'
exit $?
;;
*)
Expand Down
Loading
Loading