Skip to content

Commit

Permalink
Add the ability to determine caller origin for AnsibleTower workflows
Browse files Browse the repository at this point in the history
This is currently a mandatory feature, but will likely be optional or completely removed in the future.
Right now, there are 4 classifications of origin points:
1. Broker CLI
2. Test Function
3. Pytest Fixture
4. Unkown
  • Loading branch information
JacobCallahan authored and mshriver committed Aug 3, 2022
1 parent 747c96f commit 032ea55
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 1 deletion.
16 changes: 16 additions & 0 deletions broker/helpers.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
"""Miscellaneous helpers live here"""
import getpass
import inspect
import json
import logging
import os
Expand Down Expand Up @@ -489,3 +491,17 @@ def from_nonduplexed_exec(cls, nonduplex_exec):
stdout=nonduplex_exec.output.decode("utf-8"),
stderr="",
)


def find_origin():
"""Move up the call stack to find tests, fixtures, or cli invocations"""
prev = None
for frame in inspect.stack():
if frame.function == "checkout" and frame.filename.endswith("broker/commands.py"):
return f"broker_cli:{getpass.getuser()}"
if frame.function.startswith("test_"):
return f"{frame.function}:{frame.filename}"
if frame.function == "call_fixture_func":
return prev or "Uknown fixture"
prev = f"{frame.function}:{frame.filename}"
return f"Unknown origin by {getpass.getuser()}"
3 changes: 2 additions & 1 deletion broker/providers/ansible_tower.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
from functools import cached_property
from dynaconf import Validator
from broker import exceptions
from broker.helpers import results_filter
from broker.helpers import find_origin, results_filter
from broker.settings import settings
from logzero import logger
from datetime import datetime
Expand Down Expand Up @@ -416,6 +416,7 @@ def execute(self, **kwargs):
if name := kwargs.get("workflow"):
subject = "workflow"
get_path = self.v2.workflow_job_templates
kwargs["_broker_origin"] = find_origin()
elif name := kwargs.get("job_template"):
subject = "job_template"
get_path = self.v2.job_templates
Expand Down

0 comments on commit 032ea55

Please sign in to comment.