File pepper-0.7.6+git44.996fc29.obscpio of Package pepper
07070100000000000041ED00000000000000000000000267D9C77000000000000000000000000000000000000000000000002300000000pepper-0.7.6+git44.996fc29/.github07070100000001000041ED00000000000000000000000267D9C77000000000000000000000000000000000000000000000002D00000000pepper-0.7.6+git44.996fc29/.github/workflows07070100000002000081A400000000000000000000000167D9C77000000202000000000000000000000000000000000000003900000000pepper-0.7.6+git44.996fc29/.github/workflows/flake8.yamlname: flake8 lint
on:
push:
pull_request:
jobs:
flake8-lint:
runs-on: ubuntu-24.04
name: flake8 lint
steps:
- name: Setup python for flake8
uses: actions/setup-python@v4
with:
python-version: "3.10"
- uses: actions/checkout@v3
- name: Install nox
run: python -m pip install nox==2024.10.09
- name: Setup flake8
run: nox --force-color -e flake8 --install-only
- name: Run flake8
run: nox --force-color -e flake8 -vv
07070100000003000081A400000000000000000000000167D9C7700000093B000000000000000000000000000000000000003700000000pepper-0.7.6+git44.996fc29/.github/workflows/test.yamlname: test
on:
workflow_dispatch:
push:
pull_request:
schedule:
- cron: "0 8 * * *"
jobs:
pre-commit:
name: pre-commit
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: "3.10"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install pre-commit
pre-commit install
- name: Run pre-commit
run: pre-commit run --all-files
test:
name: test ${{ matrix.py }} - ${{ matrix.netapi }} - ${{ matrix.salt }}
runs-on: ubuntu-22.04
strategy:
fail-fast: false
matrix:
py:
- "3.8"
- "3.9"
- "3.10"
- "3.11"
netapi:
- "cherrypy"
- "tornado"
salt:
- "v3006.9"
- "v3007.1"
- "master"
exclude:
- py: "3.8"
salt: "v3007.1"
- py: "3.9"
salt: "v3007.1"
- py: "3.8"
salt: "master"
- py: "3.9"
salt: "master"
env:
SALT_REQUIREMENT: ${{ matrix.salt == 'master' && 'salt@git+https://github.com/saltstack/salt.git@master' || format('salt=={0}', matrix.salt) }}
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.py }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.py }}
- name: Install setuptools_scm
run: python -m pip install setuptools_scm
- name: Install dependencies
run: sudo apt update && sudo apt install -y libc6-dev libffi-dev gcc git openssh-server libzmq3-dev
env:
DEBIAN_FRONTEND: noninteractive
- name: Install Nox
run: |
python -m pip install --upgrade pip
pip install nox
- name: Install Test Requirements
env:
PYTHON_VERSIONS: ${{ matrix.py }}
API_BACKEND: ${{ matrix.netapi }}
run: |
nox --force-color -e tests --install-only
- name: Test
env:
PYTHON_VERSIONS: ${{ matrix.py }}
API_BACKEND: ${{ matrix.netapi }}
SKIP_REQUIREMENTS_INSTALL: YES
run: |
nox --force-color -e tests -- -vv tests/
07070100000004000081A400000000000000000000000167D9C77000000058000000000000000000000000000000000000002600000000pepper-0.7.6+git44.996fc29/.gitignore/build
*.py[co]
*.swp
*.DS_Store
MANIFEST
dist/
salt_pepper.egg-info/
.nox/
artifacts/
07070100000005000081A400000000000000000000000167D9C77000000827000000000000000000000000000000000000003300000000pepper-0.7.6+git44.996fc29/.pre-commit-config.yaml---
minimum_pre_commit_version: 3.5.0
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: check-merge-conflict # Check for files that contain merge conflict strings.
- id: trailing-whitespace # Trims trailing whitespace.
args: [--markdown-linebreak-ext=md]
- id: mixed-line-ending # Replaces or checks mixed line ending.
args: [--fix=lf]
- id: end-of-file-fixer # Makes sure files end in a newline and only a newline.
- id: check-merge-conflict # Check for files that contain merge conflict strings.
- id: check-ast # Simply check whether files parse as valid python.
# ----- Formatting ---------------------------------------------------------------------------->
- repo: https://github.com/myint/autoflake
rev: v2.2.1
hooks:
- id: autoflake
name: Remove unused variables and imports
language: python
args: ["--in-place", "--remove-all-unused-imports", "--remove-unused-variables", "--expand-star-imports"]
files: \.py$
- repo: https://github.com/saltstack/pre-commit-remove-import-headers
rev: 1.1.0
hooks:
- id: remove-import-headers
- repo: https://github.com/saltstack/salt-rewrite
# Automatically rewrite code with known rules
rev: 2.5.2
hooks:
- id: salt-rewrite
alias: rewrite-tests
name: Rewrite the test suite
files: ^tests/.*\.py$
args: [--silent, -E, fix_docstrings]
- repo: https://github.com/asottile/pyupgrade
rev: v3.15.0
hooks:
- id: pyupgrade
name: Rewrite Code to be Py3+
args: [
--py3-plus
]
- repo: https://github.com/asottile/reorder_python_imports
rev: v3.12.0
hooks:
- id: reorder-python-imports
args: [
--py3-plus,
]
- repo: https://github.com/psf/black
rev: 23.12.1
hooks:
- id: black
args: [-l 100]
# <---- Formatting -----------------------------------------------------------------------------
07070100000006000081A400000000000000000000000167D9C77000000309000000000000000000000000000000000000002700000000pepper-0.7.6+git44.996fc29/.travis.ymldist: xenial
sudo: false
services:
- docker
language: python
services:
- docker
install:
- pip install tox
python:
- '2.7'
- '3.4'
- '3.5'
- '3.6'
- '3.7'
- '3.8-dev'
env:
- SALT=v2018.3 BACKEND=cherrypy
- SALT=v2018.3 BACKEND=tornado
- SALT=v2019.2 BACKEND=cherrypy
- SALT=v2019.2 BACKEND=tornado
- SALT=develop BACKEND=cherrypy
- SALT=develop BACKEND=tornado
matrix:
allow_failures:
- python: '3.8-dev'
- env: SALT=develop BACKEND=tornado
- env: SALT=develop BACKEND=cherrypy
script:
- PYTHON="${TRAVIS_PYTHON_VERSION/-dev/-rc}"
- TOX_VERSION="py${PYTHON//./}"
- docker run -v $PWD:/pepper -ti --rm "python:$PYTHON" make -C /pepper test PYVERSION="${TOX_VERSION%-rc}" SALT="${SALT}" BACKEND="${BACKEND}"
after_success:
- sudo chown $USER .tox/
- tox -e codecov
07070100000007000081A400000000000000000000000167D9C77000000271000000000000000000000000000000000000002300000000pepper-0.7.6+git44.996fc29/LICENSE Salt - Remote execution system
Copyright 2014-2016 SaltStack Team
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
07070100000008000081A400000000000000000000000167D9C77000000023000000000000000000000000000000000000002700000000pepper-0.7.6+git44.996fc29/MANIFEST.ininclude LICENSE
include README.rst
07070100000009000081A400000000000000000000000167D9C770000000FE000000000000000000000000000000000000002400000000pepper-0.7.6+git44.996fc29/MakefilePYVERSION?=py37
SALT?=v2019.2
BACKEND?=cherrypy
DEBIAN_FRONTEND=noninteractive
install:
apt update && apt install -y libc6-dev libffi-dev gcc git openssh-server libzmq3-dev
pip install tox
test: install
tox -e flake8,$(PYVERSION)-$(BACKEND)-$(SALT)
0707010000000A000081A400000000000000000000000167D9C77000000B09000000000000000000000000000000000000002600000000pepper-0.7.6+git44.996fc29/README.rst======
Pepper
======
.. image:: https://img.shields.io/pypi/v/salt-pepper.svg
:target: https://pypi.org/project/salt-pepper
.. image:: https://travis-ci.com/saltstack/pepper.svg?branch=develop
:target: https://travis-ci.com/saltstack/pepper
.. image:: https://img.shields.io/pypi/pyversions/salt-pepper.svg
:target: https://pypi.org/project/salt-pepper
.. image:: https://img.shields.io/badge/license-Apache2-blue.svg?maxAge=3600
:target: https://pypi.org/project/salt-pepper
.. image:: https://codecov.io/gh/saltstack/pepper/branch/develop/graph/badge.svg
:target: https://codecov.io/gh/saltstack/pepper/branch/develop
Pepper contains a Python library and CLI scripts for accessing a remote
`salt-api`__ instance.
``pepperlib`` abstracts the HTTP calls to ``salt-api`` so existing Python
projects can easily integrate with a remote Salt installation just by
instantiating a class.
The ``pepper`` CLI script allows users to execute Salt commands from computers
that are external to computers running the ``salt-master`` or ``salt-minion``
daemons as though they were running Salt locally. The long-term goal is to add
additional CLI scripts maintain the same interface as Salt's own CLI scripts
(``salt``, ``salt-run``, ``salt-key``, etc).
It does not require any additional dependencies and runs on Python 2.5+ and
Python 3. (Python 3 support is new, please file an issue if you encounter
trouble.)
.. __: https://github.com/saltstack/salt-api
Installation
------------
.. code-block:: bash
pip install salt-pepper
Usage
-----
Basic usage is in heavy flux. You can run pepper using the script in %PYTHONHOME%/scripts/pepper (a pepper.cmd wrapper is provided for convenience to Windows users).
.. code-block:: bash
export SALTAPI_USER=saltdev SALTAPI_PASS=saltdev SALTAPI_EAUTH=pam
pepper '*' test.ping
pepper '*' test.kwarg hello=dolly
Examples leveraging the runner client.
.. code-block:: bash
pepper --client runner reactor.list
pepper --client runner reactor.add event='test/provision/*' reactors='/srv/salt/state/reactor/test-provision.sls'
Configuration
-------------
You can configure pepper through the command line, using environment variables
or in a configuration file ``$HOME/.pepperrc`` with the following syntax :
.. code-block::
[main]
SALTAPI_URL=https://localhost:8000/
SALTAPI_USER=saltdev
SALTAPI_PASS=saltdev
SALTAPI_EAUTH=pam
Contributing
------------
Please feel free to get involved by sending pull requests or join us on the
Salt mailing list or on IRC in #salt or #salt-devel.
This repo follows the same `contributing guidelines`__ as Salt and uses
separate develop and master branches for in-progress additions and bug-fix
changes respectively.
.. __: https://docs.saltstack.com/en/latest/topics/development/contributing.html
0707010000000B000081A400000000000000000000000167D9C77000000690000000000000000000000000000000000000002F00000000pepper-0.7.6+git44.996fc29/azure-pipelines.ymltrigger:
- develop
variables:
python: '["py2.7", "py3.4", "py3.5", "py3.6", "py3.7", "py3.8"]'
salt: '["v2018.3", "v2019.2", "develop"]'
backends: '["cherrypy", "tornado"]'
jobs:
- job: build_matrix
pool:
vmImage: 'Ubuntu-16.04'
steps:
- task: UsePythonVersion@0
inputs:
versionSpec: '3.7'
architecture: x64
- task: PythonScript@0
name: matrix
inputs:
scriptSource: Inline
script: |
import json
matrix = {}
for pyver in $(python):
for saltver in $(salt):
for backend in $(backends):
matrix['{0}-{1}-{2}'.format(pyver, backend, saltver)] = {
'python.version': pyver[2:],
'version': pyver.replace('.', ''),
'salt': saltver,
'backend': backend,
}
print('##vso[task.setvariable variable=matrix;isOutput=true]{0}'.format(json.dumps(matrix)))
- job: test_pepper
dependsOn: build_matrix
pool:
vmImage: 'Ubuntu-16.04'
strategy:
matrix: $[ dependencies.build_matrix.outputs['matrix.matrix'] ]
steps:
- task: UsePythonVersion@0
inputs:
versionSpec: '$(python.version)'
architecture: x64
- script: |
pip install tox
displayName: Install dependencies
- script: |
docker run -v $PWD:/pepper --rm "python:$(python.version)" make -C /pepper test PYTHON_VERSION=$(version) SALT=$(salt) BACKEND=$(backend)
displayName: pytest
- script: |
sudo chown $USER .tox/
tox -e codecov
displayName: codecov
env:
CODECOV_TOKEN: '16c2a232-4329-438c-b163-ccbfeeab47aa'
0707010000000C000081A400000000000000000000000167D9C77000000175000000000000000000000000000000000000002700000000pepper-0.7.6+git44.996fc29/codecov.ymlcodecov:
notify:
require_ci_to_pass: yes
branch: develop
coverage:
precision: 2
round: down
range: "50...100"
status:
project: yes
patch: yes
changes: no
parsers:
gcov:
branch_detection:
conditional: yes
loop: yes
method: no
macro: no
comment:
layout: "header, diff"
behavior: default
require_changes: no
0707010000000D000081A400000000000000000000000167D9C770000017BC000000000000000000000000000000000000002600000000pepper-0.7.6+git44.996fc29/noxfile.pyimport datetime
import os
import pathlib
import shutil
import nox
from nox.command import CommandFailed
API_BACKEND = [os.environ.get("API_BACKEND")] or ["cherrypy", "tornado"]
PYTHON_VERSIONS = [os.environ.get("PYTHON_VERSIONS")] or ["3.8", "3.9", "3.10", "3.11"]
# Nox options
# Reuse existing virtualenvs
nox.options.reuse_existing_virtualenvs = True
# Don't fail on missing interpreters
nox.options.error_on_missing_interpreters = False
# Be verbose when running under a CI context
CI_RUN = os.environ.get("CI")
PIP_INSTALL_SILENT = CI_RUN is False
SKIP_REQUIREMENTS_INSTALL = "SKIP_REQUIREMENTS_INSTALL" in os.environ
EXTRA_REQUIREMENTS_INSTALL = os.environ.get("EXTRA_REQUIREMENTS_INSTALL")
SALT_REQUIREMENT = os.environ.get("SALT_REQUIREMENT") or "salt"
COVERAGE_VERSION_REQUIREMENT = "coverage==5.5"
# Prevent Python from writing bytecode
os.environ["PYTHONDONTWRITEBYTECODE"] = "1"
# Global Path Definitions
REPO_ROOT = pathlib.Path(__file__).resolve().parent
# Change current directory to REPO_ROOT
os.chdir(str(REPO_ROOT))
ARTIFACTS_DIR = REPO_ROOT / "artifacts"
# Make sure the artifacts directory exists
ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True)
RUNTESTS_LOGFILE = ARTIFACTS_DIR / "runtests-{}.log".format(
datetime.datetime.now().strftime("%Y%m%d%H%M%S.%f")
)
COVERAGE_REPORT_DB = REPO_ROOT / ".coverage"
COVERAGE_REPORT_PROJECT = ARTIFACTS_DIR.relative_to(REPO_ROOT) / "coverage-project.xml"
COVERAGE_REPORT_TESTS = ARTIFACTS_DIR.relative_to(REPO_ROOT) / "coverage-tests.xml"
JUNIT_REPORT = ARTIFACTS_DIR.relative_to(REPO_ROOT) / "junit-report.xml"
def _install_requirements(
session,
*passed_requirements, # pylint: disable=unused-argument
install_coverage_requirements=True,
install_test_requirements=True,
install_source=True,
install_salt=True,
install_extras=None,
):
install_extras = install_extras or []
if SKIP_REQUIREMENTS_INSTALL is False:
# Always have the wheel package installed
session.install("--progress-bar=off", "wheel", silent=PIP_INSTALL_SILENT)
if install_coverage_requirements:
session.install(
"--progress-bar=off", COVERAGE_VERSION_REQUIREMENT, silent=PIP_INSTALL_SILENT
)
# Install the latest salt package
if install_salt:
session.install("--progress-bar=off", SALT_REQUIREMENT, silent=PIP_INSTALL_SILENT)
if install_test_requirements:
requirements_file = REPO_ROOT / "tests" / "requirements.txt"
install_command = [
"--progress-bar=off",
"-r",
str(requirements_file.relative_to(REPO_ROOT)),
]
session.install(*install_command, silent=PIP_INSTALL_SILENT)
if EXTRA_REQUIREMENTS_INSTALL:
session.log(
"Installing the following extra requirements because the "
"EXTRA_REQUIREMENTS_INSTALL environment variable was set: "
"EXTRA_REQUIREMENTS_INSTALL='%s'",
EXTRA_REQUIREMENTS_INSTALL,
)
install_command = ["--progress-bar=off"]
install_command += [req.strip() for req in EXTRA_REQUIREMENTS_INSTALL.split()]
session.install(*install_command, silent=PIP_INSTALL_SILENT)
if install_source:
pkg = "."
if install_extras:
pkg += f"[{','.join(install_extras)}]"
session.install("-e", pkg, silent=PIP_INSTALL_SILENT)
elif install_extras:
pkg = f".[{','.join(install_extras)}]"
session.install(pkg, silent=PIP_INSTALL_SILENT)
"""
Nox session to run tests for corresponding python versions and salt versions
"""
@nox.session(python=PYTHON_VERSIONS)
@nox.parametrize("api_backend", API_BACKEND)
def tests(session, api_backend):
_install_requirements(session)
args = [
"--rootdir",
str(REPO_ROOT),
f"--log-file={RUNTESTS_LOGFILE.relative_to(REPO_ROOT)}",
"--log-file-level=debug",
"--show-capture=no",
f"--junitxml={JUNIT_REPORT}",
"--showlocals",
"-ra",
"-s",
]
if session._runner.global_config.forcecolor:
args.append("--color=yes")
if not session.posargs:
args.append("tests/")
else:
for arg in session.posargs:
if arg.startswith("--color") and args[0].startswith("--color"):
args.pop(0)
args.append(arg)
for arg in session.posargs:
if arg.startswith("-"):
continue
if arg.startswith(f"tests{os.sep}"):
break
try:
pathlib.Path(arg).resolve().relative_to(REPO_ROOT / "tests")
break
except ValueError:
continue
else:
args.append("tests/")
try:
session.run(
"coverage", "run", "-m", "pytest", f"--salt-api-backend=rest_{api_backend}", *args
)
finally:
# Always combine and generate the XML coverage report
try:
session.run("coverage", "combine")
except CommandFailed:
# Sometimes some of the coverage files are corrupt which would
# trigger a CommandFailed exception
pass
# Generate report for salt code coverage
session.run(
"coverage",
"xml",
"-o",
str(COVERAGE_REPORT_PROJECT),
)
try:
session.run("coverage", "report", "--show-missing", "--include=pepper/*")
finally:
# Move the coverage DB to artifacts/coverage in order for it to be archived by CI
if COVERAGE_REPORT_DB.exists():
shutil.move(str(COVERAGE_REPORT_DB), str(ARTIFACTS_DIR / COVERAGE_REPORT_DB.name))
@nox.session(python="3.10")
def flake8(session):
_install_requirements(session)
# Install flake8
session.install("flake8")
# Run flake8
session.run("flake8", "tests/", "pepper/", "scripts/pepper", "setup.py")
0707010000000E000041ED00000000000000000000000267D9C77000000000000000000000000000000000000000000000002200000000pepper-0.7.6+git44.996fc29/pepper0707010000000F000081A400000000000000000000000167D9C770000001D5000000000000000000000000000000000000002E00000000pepper-0.7.6+git44.996fc29/pepper/__init__.py"""
Pepper is a CLI front-end to salt-api
"""
from importlib.metadata import PackageNotFoundError
from importlib.metadata import version
from pepper.libpepper import Pepper
from pepper.libpepper import PepperException
__all__ = ("__version__", "Pepper", "PepperException")
try:
__version__ = version("salt_pepper")
except PackageNotFoundError:
# package is not installed
__version__ = None
# For backwards compatibility
version = __version__
sha = None
07070100000010000081A400000000000000000000000167D9C77000006396000000000000000000000000000000000000002900000000pepper-0.7.6+git44.996fc29/pepper/cli.py"""
A CLI interface to a remote salt-api instance
"""
import getpass
import json
import logging
import optparse
import os
import sys
import textwrap
import time
import pepper
from pepper.exceptions import PepperArgumentsException
from pepper.exceptions import PepperAuthException
from pepper.exceptions import PepperException
# Import Pepper Libraries
try:
# Python 3
from configparser import ConfigParser, RawConfigParser
except ImportError:
# Python 2
from ConfigParser import ConfigParser, RawConfigParser
try:
# Python 3
JSONDecodeError = json.decode.JSONDecodeError
except AttributeError:
# Python 2
JSONDecodeError = ValueError
try:
input = raw_input
except NameError:
pass
if sys.version_info[0] == 2:
FileNotFoundError = IOError
logger = logging.getLogger(__name__)
class PepperCli:
def __init__(self, seconds_to_wait=3):
self.seconds_to_wait = seconds_to_wait
self.parser = self.get_parser()
self.parser.option_groups.extend(
[
self.add_globalopts(),
self.add_tgtopts(),
self.add_authopts(),
self.add_retcodeopts(),
]
)
self.parse()
def get_parser(self):
return optparse.OptionParser(
description=__doc__, usage="%prog [opts]", version=pepper.__version__
)
def parse(self):
"""
Parse all args
"""
self.parser.add_option(
"-c",
dest="config",
default=os.environ.get("PEPPERRC", os.path.join(os.path.expanduser("~"), ".pepperrc")),
help=textwrap.dedent(
"""
Configuration file location. Default is a file path in the
"PEPPERRC" environment variable or ~/.pepperrc.
"""
),
)
self.parser.add_option(
"-p",
dest="profile",
default=os.environ.get("PEPPERPROFILE", "main"),
help=textwrap.dedent(
"""
Profile in config file to use. Default is "PEPPERPROFILE" environment
variable or 'main'
"""
),
)
self.parser.add_option(
"-m",
dest="master",
default=os.environ.get(
"MASTER_CONFIG",
os.path.join(os.path.expanduser("~"), ".config", "pepper", "master"),
),
help=textwrap.dedent(
"""
Master Configuration file location for configuring outputters.
default: ~/.config/pepper/master
"""
),
)
self.parser.add_option(
"-o",
"--out",
dest="output",
default=None,
help=textwrap.dedent(
"""
Salt outputter to use for printing out returns.
"""
),
)
self.parser.add_option(
"--output-file",
dest="output_file",
default=None,
help=textwrap.dedent(
"""
File to put command output in
"""
),
)
self.parser.add_option(
"-v",
dest="verbose",
default=0,
action="count",
help=textwrap.dedent(
"""
Increment output verbosity; may be specified multiple times
"""
),
)
self.parser.add_option(
"-H",
"--debug-http",
dest="debug_http",
default=False,
action="store_true",
help=textwrap.dedent(
"""
Output the HTTP request/response headers on stderr
"""
),
)
self.parser.add_option(
"--ignore-ssl-errors",
action="store_true",
dest="ignore_ssl_certificate_errors",
default=False,
help=textwrap.dedent(
"""
Ignore any SSL certificate that may be encountered. Note that it is
recommended to resolve certificate errors for production.
"""
),
)
self.options, self.args = self.parser.parse_args()
option_names = ["fail_any", "fail_any_none", "fail_all", "fail_all_none"]
toggled_options = [name for name in option_names if getattr(self.options, name)]
if len(toggled_options) > 1:
s = repr(toggled_options).strip("[]")
self.parser.error("Options %s are mutually exclusive" % s)
def add_globalopts(self):
"""
Misc global options
"""
optgroup = optparse.OptionGroup(
self.parser, "Pepper ``salt`` Options", "Mimic the ``salt`` CLI"
)
optgroup.add_option(
"-t",
"--timeout",
dest="timeout",
type="int",
default=60,
help=textwrap.dedent(
"""
Specify wait time (in seconds) before returning control to the shell
"""
),
)
optgroup.add_option(
"--client",
dest="client",
default="local",
help=textwrap.dedent(
"""
specify the salt-api client to use (local, local_async,
runner, etc)
"""
),
)
optgroup.add_option(
"--json",
dest="json_input",
help=textwrap.dedent(
"""
Enter JSON at the CLI instead of positional (text) arguments. This
is useful for arguments that need complex data structures.
Specifying this argument will cause positional arguments to be
ignored.
"""
),
)
optgroup.add_option(
"--json-file",
dest="json_file",
help=textwrap.dedent(
"""
Specify file containing the JSON to be used by pepper
"""
),
)
optgroup.add_option(
"--fail-if-incomplete",
action="store_true",
dest="fail_if_minions_dont_respond",
default=False,
help=textwrap.dedent(
"""
Return a failure exit code if not all minions respond. This option
requires the authenticated user have access to run the
`jobs.list_jobs` runner function.
"""
),
)
return optgroup
def add_tgtopts(self):
"""
Targeting
"""
optgroup = optparse.OptionGroup(
self.parser, "Targeting Options", "Target which minions to run commands on"
)
optgroup.defaults.update({"expr_form": "glob"})
optgroup.add_option(
"-E",
"--pcre",
dest="expr_form",
action="store_const",
const="pcre",
help="Target hostnames using PCRE regular expressions",
)
optgroup.add_option(
"-L",
"--list",
dest="expr_form",
action="store_const",
const="list",
help="Specify a comma delimited list of hostnames",
)
optgroup.add_option(
"-G",
"--grain",
dest="expr_form",
action="store_const",
const="grain",
help="Target based on system properties",
)
optgroup.add_option(
"--grain-pcre",
dest="expr_form",
action="store_const",
const="grain_pcre",
help="Target based on PCRE matches on system properties",
)
optgroup.add_option(
"-I",
"--pillar",
dest="expr_form",
action="store_const",
const="pillar",
help="Target based on pillar values",
)
optgroup.add_option(
"--pillar-pcre",
dest="expr_form",
action="store_const",
const="pillar_pcre",
help="Target based on PCRE matches on pillar values",
)
optgroup.add_option(
"-R",
"--range",
dest="expr_form",
action="store_const",
const="range",
help="Target based on range expression",
)
optgroup.add_option(
"-C",
"--compound",
dest="expr_form",
action="store_const",
const="compound",
help="Target based on compound expression",
)
optgroup.add_option(
"-N",
"--nodegroup",
dest="expr_form",
action="store_const",
const="nodegroup",
help="Target based on a named nodegroup",
)
optgroup.add_option("--batch", dest="batch", default=None)
return optgroup
def add_authopts(self):
"""
Authentication options
"""
optgroup = optparse.OptionGroup(
self.parser,
"Authentication Options",
textwrap.dedent(
"""
Authentication credentials can optionally be supplied via the
environment variables:
SALTAPI_URL, SALTAPI_USER, SALTAPI_PASS, SALTAPI_EAUTH.
"""
),
)
optgroup.add_option(
"-u",
"--saltapi-url",
dest="saltapiurl",
help="Specify the host url. Defaults to https://localhost:8080",
)
optgroup.add_option(
"-a",
"--auth",
"--eauth",
"--extended-auth",
dest="eauth",
help=textwrap.dedent(
"""
Specify the external_auth backend to authenticate against and
interactively prompt for credentials
"""
),
)
optgroup.add_option(
"--username",
dest="username",
help=textwrap.dedent(
"""
Optional, defaults to user name. will be prompt if empty unless --non-interactive
"""
),
)
optgroup.add_option(
"--password",
dest="password",
help=textwrap.dedent(
"""
Optional, but will be prompted unless --non-interactive
"""
),
)
optgroup.add_option(
"--token-expire",
dest="token_expire",
help=textwrap.dedent(
"""
Set eauth token expiry in seconds. Must be allowed per
user. See the `token_expire_user_override` Master setting
for more info.
"""
),
)
optgroup.add_option(
"--non-interactive",
action="store_false",
dest="interactive",
default=True,
help=textwrap.dedent(
"""
Optional, fail rather than waiting for input
"""
),
)
optgroup.add_option(
"-T",
"--make-token",
default=False,
dest="mktoken",
action="store_true",
help=textwrap.dedent(
"""
Generate and save an authentication token for re-use. The token is
generated and made available for the period defined in the Salt
Master.
"""
),
)
optgroup.add_option(
"-r",
"--run-uri",
default=False,
dest="userun",
action="store_true",
help=textwrap.dedent(
"""
Use an eauth token from /token and send commands through the
/run URL instead of the traditional session token
approach.
"""
),
)
optgroup.add_option(
"-x",
dest="cache",
default=os.environ.get(
"PEPPERCACHE", os.path.join(os.path.expanduser("~"), ".peppercache")
),
help=textwrap.dedent(
"""
Cache file location. Default is a file path in the
"PEPPERCACHE" environment variable or ~/.peppercache.
"""
),
)
return optgroup
def add_retcodeopts(self):
"""
ret code validation options
"""
optgroup = optparse.OptionGroup(
self.parser, "retcode Field Validation Options", "Validate return.HOST.retcode fields"
)
optgroup.add_option(
"--fail-any",
dest="fail_any",
action="store_true",
help="Fail if any of retcode field is non zero.",
)
optgroup.add_option(
"--fail-any-none",
dest="fail_any_none",
action="store_true",
help="Fail if any of retcode field is non zero or there is no retcode at all.",
)
optgroup.add_option(
"--fail-all",
dest="fail_all",
action="store_true",
help="Fail if all retcode fields are non zero.",
)
optgroup.add_option(
"--fail-all-none",
dest="fail_all_none",
action="store_true",
help="Fail if all retcode fields are non zero or there is no retcode at all.",
)
return optgroup
def get_login_details(self):
"""
This parses the config file, environment variables and command line options
and returns the config values
Order of parsing:
command line options, ~/.pepperrc, environment, defaults
"""
# setting default values
results = {
"SALTAPI_USER": None,
"SALTAPI_PASS": None,
"SALTAPI_EAUTH": "auto",
}
try:
config = ConfigParser(interpolation=None)
except TypeError:
config = RawConfigParser()
config.read(self.options.config)
# read file
profile = self.options.profile
if config.has_section(profile):
for key, value in list(results.items()):
if config.has_option(profile, key):
results[key] = config.get(profile, key)
# get environment values
for key, value in list(results.items()):
results[key] = os.environ.get(key, results[key])
if results["SALTAPI_EAUTH"] == "kerberos":
results["SALTAPI_PASS"] = None
if self.options.eauth:
results["SALTAPI_EAUTH"] = self.options.eauth
if self.options.token_expire:
results["SALTAPI_TOKEN_EXPIRE"] = self.options.token_expire
if self.options.username is None and results["SALTAPI_USER"] is None:
if self.options.interactive:
results["SALTAPI_USER"] = input("Username: ")
else:
raise PepperAuthException("SALTAPI_USER required")
else:
if self.options.username is not None:
results["SALTAPI_USER"] = self.options.username
if (
self.options.password is None
and results["SALTAPI_PASS"] is None
and results["SALTAPI_EAUTH"] != "kerberos"
):
if self.options.interactive:
results["SALTAPI_PASS"] = getpass.getpass(prompt="Password: ")
else:
raise PepperAuthException("SALTAPI_PASS required")
else:
if self.options.password is not None:
results["SALTAPI_PASS"] = self.options.password
return results
def parse_url(self):
"""
Determine api url
"""
url = "https://localhost:8000/"
try:
config = ConfigParser(interpolation=None)
except TypeError:
config = RawConfigParser()
config.read(self.options.config)
# read file
profile = self.options.profile
if config.has_section(profile):
if config.has_option(profile, "SALTAPI_URL"):
url = config.get(profile, "SALTAPI_URL")
# get environment values
url = os.environ.get("SALTAPI_URL", url)
# get eauth prompt options
if self.options.saltapiurl:
url = self.options.saltapiurl
return url
def parse_login(self):
"""
Extract the authentication credentials
"""
login_details = self.get_login_details()
# Auth values placeholder; grab interactively at CLI or from config
username = login_details["SALTAPI_USER"]
password = login_details["SALTAPI_PASS"]
eauth = login_details["SALTAPI_EAUTH"]
ret = dict(username=username, password=password, eauth=eauth)
token_expire = login_details.get("SALTAPI_TOKEN_EXPIRE", None)
if token_expire:
ret["token_expire"] = int(token_expire)
return ret
def parse_cmd(self, api):
"""
Extract the low data for a command from the passed CLI params
"""
# Short-circuit if JSON was given.
if self.options.json_input:
try:
return json.loads(self.options.json_input)
except JSONDecodeError:
raise PepperArgumentsException("Invalid JSON given.")
if self.options.json_file:
try:
with open(self.options.json_file) as json_content:
try:
return json.load(json_content)
except JSONDecodeError:
raise PepperArgumentsException("Invalid JSON given.")
except FileNotFoundError:
raise PepperArgumentsException("Cannot open file: %s", self.options.json_file)
args = list(self.args)
client = self.options.client if not self.options.batch else "local_batch"
low = {"client": client}
if client.startswith("local"):
if len(args) < 2:
self.parser.error("Command or target not specified")
low["tgt_type"] = self.options.expr_form
low["tgt"] = args.pop(0)
low["fun"] = args.pop(0)
low["batch"] = self.options.batch
low["arg"] = args
elif client.startswith("runner"):
low["fun"] = args.pop(0)
# post https://github.com/saltstack/salt/pull/50124, kwargs can be
# passed as is in foo=bar form, splitting and deserializing will
# happen in salt-api. additionally, the presence of salt-version header
# means we are neon or newer, so don't need a finer grained check
if api.salt_version:
low["arg"] = args
else:
for arg in args:
if "=" in arg:
key, value = arg.split("=", 1)
try:
low[key] = json.loads(value)
except JSONDecodeError:
low[key] = value
else:
low.setdefault("arg", []).append(arg)
elif client.startswith("wheel"):
low["fun"] = args.pop(0)
# see above comment in runner arg handling
if api.salt_version:
low["arg"] = args
else:
for arg in args:
if "=" in arg:
key, value = arg.split("=", 1)
try:
low[key] = json.loads(value)
except JSONDecodeError:
low[key] = value
else:
low.setdefault("arg", []).append(arg)
elif client.startswith("ssh"):
if len(args) < 2:
self.parser.error("Command or target not specified")
low["tgt_type"] = self.options.expr_form
low["tgt"] = args.pop(0)
low["fun"] = args.pop(0)
low["batch"] = self.options.batch
low["arg"] = args
else:
raise PepperException("Client not implemented: {}".format(client))
return [low]
def poll_for_returns(self, api, load):
"""
Run a command with the local_async client and periodically poll the job
cache for returns for the job.
"""
load[0]["client"] = "local_async"
async_ret = self.low(api, load)
jid = async_ret["return"][0]["jid"]
nodes = async_ret["return"][0]["minions"]
ret_nodes = []
exit_code = 1
# keep trying until all expected nodes return
total_time = 0
start_time = time.time()
exit_code = 0
while True:
total_time = time.time() - start_time
if total_time > self.options.timeout:
exit_code = 1
break
jid_ret = self.low(
api,
[
{
"client": "runner",
"fun": "jobs.lookup_jid",
"kwarg": {
"jid": jid,
},
}
],
)
inner_ret = jid_ret["return"][0]
# sometimes ret is nested in data
if "data" in inner_ret:
inner_ret = inner_ret["data"]
responded = set(inner_ret.keys()) ^ set(ret_nodes)
for node in responded:
yield None, [{node: inner_ret[node]}]
ret_nodes = list(inner_ret.keys())
if set(ret_nodes) == set(nodes):
exit_code = 0
break
else:
time.sleep(self.seconds_to_wait)
exit_code = exit_code if self.options.fail_if_minions_dont_respond else 0
failed = list(set(ret_nodes) ^ set(nodes))
if failed:
yield exit_code, [{"Failed": failed}]
def login(self, api):
login = api.token if self.options.userun else api.login
if self.options.mktoken:
token_file = self.options.cache
try:
with open(token_file) as f:
auth = json.load(f)
if auth["expire"] < time.time() + 30:
logger.error("Login token expired")
raise Exception("Login token expired")
except Exception as e:
if e.args[0] != 2:
logger.error("Unable to load login token from {} {}".format(token_file, str(e)))
if os.path.isfile(token_file):
os.remove(token_file)
auth = login(**self.parse_login())
try:
oldumask = os.umask(0)
fdsc = os.open(token_file, os.O_WRONLY | os.O_CREAT, 0o600)
with os.fdopen(fdsc, "wt") as f:
json.dump(auth, f)
except Exception as e:
logger.error("Unable to save token to {} {}".format(token_file, str(e)))
finally:
os.umask(oldumask)
else:
auth = login(**self.parse_login())
api.auth = auth
self.auth = auth
return auth
def low(self, api, load):
path = "/run" if self.options.userun else "/"
if self.options.userun:
for i in load:
i["token"] = self.auth["token"]
# having a defined salt_version means changes from https://github.com/saltstack/salt/pull/51979
# are available if backend is tornado, so safe to supply timeout
if self.options.timeout and api.salt_version:
for i in load:
if not i.get("client", "").startswith("wheel"):
i["timeout"] = self.options.timeout
return api.low(load, path=path)
def run(self):
"""
Parse all arguments and call salt-api
"""
# set up logging
rootLogger = logging.getLogger(name=None)
rootLogger.addHandler(logging.StreamHandler())
rootLogger.setLevel(max(logging.ERROR - (self.options.verbose * 10), 1))
api = pepper.Pepper(
self.parse_url(),
debug_http=self.options.debug_http,
ignore_ssl_errors=self.options.ignore_ssl_certificate_errors,
)
self.login(api)
load = self.parse_cmd(api)
for entry in load:
if not entry.get("client", "").startswith("wheel"):
entry["full_return"] = True
if self.options.fail_if_minions_dont_respond:
for exit_code, ret in self.poll_for_returns(api, load): # pragma: no cover
yield exit_code, json.dumps(ret, sort_keys=True, indent=4)
else:
ret = self.low(api, load)
exit_code = 0
yield exit_code, json.dumps(ret, sort_keys=True, indent=4)
07070100000011000081A400000000000000000000000167D9C77000000092000000000000000000000000000000000000003000000000pepper-0.7.6+git44.996fc29/pepper/exceptions.pyclass PepperAuthException(Exception):
pass
class PepperArgumentsException(Exception):
pass
class PepperException(Exception):
pass
07070100000012000081A400000000000000000000000167D9C77000003AB8000000000000000000000000000000000000002F00000000pepper-0.7.6+git44.996fc29/pepper/libpepper.py"""
A Python library for working with Salt's REST API
(Specifically the rest_cherrypy netapi module.)
"""
import json
import logging
import re
import ssl
from pepper.exceptions import PepperException
try:
ssl._create_default_https_context = ssl._create_stdlib_context
except Exception:
pass
try:
from urllib.request import (
HTTPHandler,
HTTPSHandler,
Request,
urlopen,
install_opener,
build_opener,
)
from urllib.error import HTTPError, URLError
import urllib.parse as urlparse
except ImportError:
from urllib2 import (
HTTPHandler,
HTTPSHandler,
Request,
urlopen,
install_opener,
build_opener,
HTTPError,
URLError,
)
import urlparse
logger = logging.getLogger(__name__)
class Pepper:
"""
A thin wrapper for making HTTP calls to the salt-api rest_cherrpy REST
interface
>>> api = Pepper('https://localhost:8000')
>>> api.login('saltdev', 'saltdev', 'pam')
{"return": [
{
"eauth": "pam",
"expire": 1370434219.714091,
"perms": [
"test.*"
],
"start": 1370391019.71409,
"token": "c02a6f4397b5496ba06b70ae5fd1f2ab75de9237",
"user": "saltdev"
}
]
}
>>> api.low([{'client': 'local', 'tgt': '*', 'fun': 'test.ping'}])
{u'return': [{u'ms-0': True,
u'ms-1': True,
u'ms-2': True,
u'ms-3': True,
u'ms-4': True}]}
"""
def __init__(self, api_url="https://localhost:8000", debug_http=False, ignore_ssl_errors=False):
"""
Initialize the class with the URL of the API
:param api_url: Host or IP address of the salt-api URL;
include the port number
:param debug_http: Add a flag to urllib2 to output the HTTP exchange
:param ignore_ssl_errors: Add a flag to urllib2 to ignore invalid SSL certificates
:raises PepperException: if the api_url is misformed
"""
split = urlparse.urlsplit(api_url)
if split.scheme not in ["http", "https"]:
raise PepperException("salt-api URL missing HTTP(s) protocol: {}".format(api_url))
self.api_url = api_url
self.debug_http = int(debug_http)
self._ssl_verify = not ignore_ssl_errors
self.auth = {}
self.salt_version = None
def req_stream(self, path):
"""
A thin wrapper to get a response from saltstack api.
The body of the response will not be downloaded immediately.
Make sure to close the connection after use.
api = Pepper('http://ipaddress/api/')
print(api.login('salt','salt','pam'))
response = api.req_stream('/events')
:param path: The path to the salt api resource
:return: :class:`Response <Response>` object
:rtype: requests.Response
"""
import requests
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
"X-Requested-With": "XMLHttpRequest",
}
if self.auth and "token" in self.auth and self.auth["token"]:
headers.setdefault("X-Auth-Token", self.auth["token"])
else:
raise PepperException("Authentication required")
return
params = {
"url": self._construct_url(path),
"headers": headers,
"verify": self._ssl_verify is True,
"stream": True,
}
try:
resp = requests.get(**params)
if resp.status_code == 401:
raise PepperException(str(resp.status_code) + ":Authentication denied")
return
if resp.status_code == 500:
raise PepperException(str(resp.status_code) + ":Server error.")
return
if resp.status_code == 404:
raise PepperException(str(resp.status_code) + " :This request returns nothing.")
return
except PepperException as e:
print(e)
return
return resp
def req_get(self, path):
"""
A thin wrapper from get http method of saltstack api
api = Pepper('http://ipaddress/api/')
print(api.login('salt','salt','pam'))
print(api.req_get('/keys'))
"""
import requests
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
"X-Requested-With": "XMLHttpRequest",
}
if self.auth and "token" in self.auth and self.auth["token"]:
headers.setdefault("X-Auth-Token", self.auth["token"])
else:
raise PepperException("Authentication required")
return
params = {
"url": self._construct_url(path),
"headers": headers,
"verify": self._ssl_verify is True,
}
try:
resp = requests.get(**params)
if resp.status_code == 401:
raise PepperException(str(resp.status_code) + ":Authentication denied")
return
if resp.status_code == 500:
raise PepperException(str(resp.status_code) + ":Server error.")
return
if resp.status_code == 404:
raise PepperException(str(resp.status_code) + " :This request returns nothing.")
return
except PepperException as e:
print(e)
return
return resp.json()
def req(self, path, data=None):
"""
A thin wrapper around urllib2 to send requests and return the response
If the current instance contains an authentication token it will be
attached to the request as a custom header.
:rtype: dictionary
"""
if (hasattr(data, "get") and data.get("eauth") == "kerberos") or self.auth.get(
"eauth"
) == "kerberos":
return self.req_requests(path, data)
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
"X-Requested-With": "XMLHttpRequest",
}
opener = build_opener()
for handler in opener.handlers:
if isinstance(handler, HTTPHandler):
handler.set_http_debuglevel(self.debug_http)
if isinstance(handler, HTTPSHandler):
handler.set_http_debuglevel(self.debug_http)
install_opener(opener)
# Build POST data
if data is not None:
postdata = json.dumps(data).encode()
clen = len(postdata)
else:
postdata = None
# Create request object
url = self._construct_url(path)
req = Request(url, postdata, headers)
# Add POST data to request
if data is not None:
req.add_header("Content-Length", clen)
# Add auth header to request
if path != "/run" and self.auth and "token" in self.auth and self.auth["token"]:
req.add_header("X-Auth-Token", self.auth["token"])
# Send request
try:
if not (self._ssl_verify):
con = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
f = urlopen(req, context=con)
else:
f = urlopen(req)
content = f.read().decode("utf-8")
if self.debug_http:
logger.debug("Response: %s", content)
ret = json.loads(content)
if not self.salt_version and "x-salt-version" in f.headers:
self._parse_salt_version(f.headers["x-salt-version"])
except (HTTPError, URLError) as exc:
logger.debug("Error with request", exc_info=True)
status = getattr(exc, "code", None)
if status == 401:
raise PepperException("Authentication denied")
if status == 500:
raise PepperException("Server error.")
logger.error("Error with request: {}".format(exc))
raise
except AttributeError:
logger.debug("Error converting response from JSON", exc_info=True)
raise PepperException("Unable to parse the server response.")
return ret
def req_requests(self, path, data=None):
"""
A thin wrapper around request and request_kerberos to send
requests and return the response
If the current instance contains an authentication token it will be
attached to the request as a custom header.
:rtype: dictionary
"""
import requests
from requests_gssapi import HTTPSPNEGOAuth, OPTIONAL
auth = HTTPSPNEGOAuth(mutual_authentication=OPTIONAL)
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
"X-Requested-With": "XMLHttpRequest",
}
if self.auth and "token" in self.auth and self.auth["token"]:
headers.setdefault("X-Auth-Token", self.auth["token"])
# Optionally toggle SSL verification
params = {
"url": self._construct_url(path),
"headers": headers,
"verify": self._ssl_verify is True,
"auth": auth,
"data": json.dumps(data),
}
logger.debug("postdata {}".format(params))
resp = requests.post(**params)
if resp.status_code == 401:
# TODO should be resp.raise_from_status
raise PepperException("Authentication denied")
if resp.status_code == 500:
# TODO should be resp.raise_from_status
raise PepperException("Server error.")
if not self.salt_version and "x-salt-version" in resp.headers:
self._parse_salt_version(resp.headers["x-salt-version"])
return resp.json()
def low(self, lowstate, path="/"):
"""
Execute a command through salt-api and return the response
:param string path: URL path to be joined with the API hostname
:param list lowstate: a list of lowstate dictionaries
"""
return self.req(path, lowstate)
def local(self, tgt, fun, arg=None, kwarg=None, expr_form="glob", timeout=None, ret=None):
"""
Run a single command using the ``local`` client
Wraps :meth:`low`.
"""
low = {
"client": "local",
"tgt": tgt,
"fun": fun,
}
if arg:
low["arg"] = arg
if kwarg:
low["kwarg"] = kwarg
if expr_form:
low["expr_form"] = expr_form
if timeout:
low["timeout"] = timeout
if ret:
low["ret"] = ret
return self.low([low])
def local_async(self, tgt, fun, arg=None, kwarg=None, expr_form="glob", timeout=None, ret=None):
"""
Run a single command using the ``local_async`` client
Wraps :meth:`low`.
"""
low = {
"client": "local_async",
"tgt": tgt,
"fun": fun,
}
if arg:
low["arg"] = arg
if kwarg:
low["kwarg"] = kwarg
if expr_form:
low["expr_form"] = expr_form
if timeout:
low["timeout"] = timeout
if ret:
low["ret"] = ret
return self.low([low])
def local_batch(self, tgt, fun, arg=None, kwarg=None, expr_form="glob", batch="50%", ret=None):
"""
Run a single command using the ``local_batch`` client
Wraps :meth:`low`.
"""
low = {
"client": "local_batch",
"tgt": tgt,
"fun": fun,
}
if arg:
low["arg"] = arg
if kwarg:
low["kwarg"] = kwarg
if expr_form:
low["expr_form"] = expr_form
if batch:
low["batch"] = batch
if ret:
low["ret"] = ret
return self.low([low])
def lookup_jid(self, jid):
"""
Get job results
Wraps :meth:`runner`.
"""
return self.runner("jobs.lookup_jid", jid="{}".format(jid))
def runner(self, fun, arg=None, **kwargs):
"""
Run a single command using the ``runner`` client
Usage::
runner('jobs.lookup_jid', jid=12345)
"""
low = {
"client": "runner",
"fun": fun,
}
if arg:
low["arg"] = arg
low.update(kwargs)
return self.low([low])
def wheel(self, fun, arg=None, kwarg=None, **kwargs):
"""
Run a single command using the ``wheel`` client
Usage::
wheel('key.accept', match='myminion')
"""
low = {
"client": "wheel",
"fun": fun,
}
if arg:
low["arg"] = arg
if kwarg:
low["kwarg"] = kwarg
low.update(kwargs)
return self.low([low])
def _send_auth(self, path, **kwargs):
return self.req(path, kwargs)
def login(self, username=None, password=None, eauth=None, **kwargs):
"""
Authenticate with salt-api and return the user permissions and
authentication token or an empty dict
"""
local = locals()
kwargs.update(
{
key: local[key]
for key in ("username", "password", "eauth")
if local.get(key, None) is not None
}
)
self.auth = self._send_auth("/login", **kwargs).get("return", [{}])[0]
return self.auth
def token(self, **kwargs):
"""
Get an eauth token from Salt for use with the /run URL
"""
self.auth = self._send_auth("/token", **kwargs)[0]
return self.auth
def _construct_url(self, path):
"""
Construct the url to salt-api for the given path
Args:
path: the path to the salt-api resource
>>> api = Pepper('https://localhost:8000/salt-api/')
>>> api._construct_url('/login')
'https://localhost:8000/salt-api/login'
"""
relative_path = path.lstrip("/")
return urlparse.urljoin(self.api_url, relative_path)
def _parse_salt_version(self, version):
# borrow from salt.version
git_describe_regex = re.compile(
r"(?:[^\d]+)?(?P<major>[\d]{1,4})"
r"\.(?P<minor>[\d]{1,2})"
r"(?:\.(?P<bugfix>[\d]{0,2}))?"
r"(?:\.(?P<mbugfix>[\d]{0,2}))?"
r"(?:(?P<pre_type>rc|a|b|alpha|beta|nb)(?P<pre_num>[\d]{1}))?"
r"(?:(?:.*)-(?P<noc>(?:[\d]+|n/a))-(?P<sha>[a-z0-9]{8}))?"
)
match = git_describe_regex.match(version)
if match:
self.salt_version = match.groups()
07070100000013000081A400000000000000000000000167D9C77000000FD7000000000000000000000000000000000000002D00000000pepper-0.7.6+git44.996fc29/pepper/retcode.py"""
A retcode validator
"""
class PepperRetcode:
"""
Validation container
"""
def validate(self, options, result):
"""
Validate result dictionary retcode values.
:param options: optparse options
:param result: dictionary from Saltstack master
:return: exit code
"""
if options.fail_any:
return self.validate_fail_any(result)
if options.fail_any_none:
return self.validate_fail_any_none(result)
if options.fail_all:
return self.validate_fail_all(result)
if options.fail_all_none:
return self.validate_fail_all_none(result)
return 0
@staticmethod
def validate_fail_any(result):
"""
Validate result dictionary retcode values.
Returns 0 if no retcode keys.
Returns first non zero retcode if any of recodes is non zero.
:param result: dictionary from Saltstack master
:return: exit code
"""
if isinstance(result, list):
if isinstance(result[0], dict):
minion = result[0]
retcodes = list(
minion[name].get("retcode")
for name in minion
if isinstance(minion[name], dict) and minion[name].get("retcode") is not None
)
return next((r for r in retcodes if r != 0), 0)
return 0
@staticmethod
def validate_fail_any_none(result):
"""
Validate result dictionary retcode values.
Returns -1 if no retcode keys.
Returns first non zero retcode if any of recodes is non zero.
:param result: dictionary from Saltstack master
:return: exit code
"""
if isinstance(result, list):
if isinstance(result[0], dict):
minion = result[0]
retcodes = list(
minion[name].get("retcode")
for name in minion
if isinstance(minion[name], dict) and minion[name].get("retcode") is not None
)
if not retcodes:
return -1 # there are no retcodes
return next((r for r in retcodes if r != 0), 0)
return -1
@staticmethod
def validate_fail_all(result):
"""
Validate result dictionary retcode values.
Returns 0 if no retcode keys.
Returns first non zero retcode if all recodes are non zero.
:param result: dictionary from Saltstack master
:return: exit code
"""
if isinstance(result, list):
if isinstance(result[0], dict):
minion = result[0]
retcodes = list(
minion[name].get("retcode")
for name in minion
if isinstance(minion[name], dict) and minion[name].get("retcode") is not None
)
if all(r != 0 for r in retcodes):
return next((r for r in retcodes if r != 0), 0)
return 0
@staticmethod
def validate_fail_all_none(result):
"""
Validate result dictionary retcode values.
Returns -1 if no retcode keys.
Returns first non zero retcode if all recodes are non zero.
:param result: dictionary from Saltstack master
:return: exit code
"""
if isinstance(result, list):
if isinstance(result[0], dict):
minion = result[0]
retcodes = list(
minion[name].get("retcode")
for name in minion
if isinstance(minion[name], dict) and minion[name].get("retcode") is not None
)
if not retcodes:
return -1 # there are no retcodes
if all(r != 0 for r in retcodes):
return next((r for r in retcodes if r != 0), 0)
else:
return 0
return -1
07070100000014000081ED00000000000000000000000167D9C770000016C0000000000000000000000000000000000000002C00000000pepper-0.7.6+git44.996fc29/pepper/script.py#!/usr/bin/env python
"""
A CLI interface to a remote salt-api instance
"""
import json
import logging
import sys
from pepper.cli import PepperCli
from pepper.exceptions import PepperArgumentsException
from pepper.exceptions import PepperAuthException
from pepper.exceptions import PepperException
from pepper.retcode import PepperRetcode
try:
import salt.loader
import salt.config
import salt.output
HAS_SALT = True
except ImportError:
HAS_SALT = False
logger = logging.getLogger(__name__)
class Pepper:
def __init__(self):
self.cli = PepperCli()
if HAS_SALT:
self.opts = salt.config.client_config(self.cli.options.master)
else:
self.opts = {}
if self.cli.options.output_file is not None:
self.opts["output_file"] = self.cli.options.output_file
@property
def output(self):
if not hasattr(self, "modules"):
self.modules = salt.loader.minion_mods(self.opts)
try:
oput = self.modules[self.cli.args[1]].__outputter__
except (KeyError, AttributeError, TypeError):
oput = "nested"
return oput
def __call__(self):
try:
for exit_code, result in self.cli.run():
if HAS_SALT and self.opts:
logger.debug("Use Salt outputters")
result = json.loads(result)
# unwrap ret in some cases
if "return" in result:
result = result["return"]
for ret in result:
if isinstance(ret, dict):
if self.cli.options.client.startswith("local"):
for minionid, minionret in ret.items():
# rest_tornado doesnt return full_return directly
# it will always be from get_event, so the output differs slightly
if isinstance(minionret, dict) and "return" in minionret:
# version >= 2017.7
salt.output.display_output(
{minionid: minionret["return"]},
self.cli.options.output
or minionret.get("out", None)
or "nested",
opts=self.opts,
)
# cherrypy returns with ret via full_return
elif isinstance(minionret, dict) and "ret" in minionret:
# version >= 2017.7
salt.output.display_output(
{minionid: minionret["ret"]},
self.cli.options.output
or minionret.get("out", None)
or "nested",
opts=self.opts,
)
else:
salt.output.display_output(
{minionid: minionret},
self.cli.options.output or self.output,
opts=self.opts,
)
elif "data" in ret:
# unfold runners
outputter = ret.get("outputter", "nested")
if isinstance(ret["data"], dict) and "return" in ret["data"]:
ret = ret["data"]["return"]
salt.output.display_output(
ret, self.cli.options.output or outputter, opts=self.opts
)
else:
salt.output.display_output(
{self.cli.options.client: ret},
self.cli.options.output or ret.get("outputter", "nested"),
opts=self.opts,
)
else:
salt.output.display_output(
{self.cli.options.client: ret},
self.cli.options.output or "nested",
opts=self.opts,
)
else:
if self.cli.options.output_file is not None:
with open(self.cli.options.output_file, "a") as ofile:
print(result, file=ofile)
else:
print(result)
if exit_code is not None:
if exit_code == 0:
return PepperRetcode().validate(self.cli.options, result)
return exit_code
except (PepperException, PepperAuthException, PepperArgumentsException) as exc:
print("Pepper error: {}".format(exc), file=sys.stderr)
return 1
except KeyboardInterrupt:
# TODO: mimic CLI and output JID on ctrl-c
return 0
except Exception as e:
print(e)
print(
"Uncaught Pepper error (increase verbosity for the full traceback).",
file=sys.stderr,
)
logger.debug("Uncaught traceback:", exc_info=True)
return 1
07070100000015000081A400000000000000000000000167D9C770000000E8000000000000000000000000000000000000002A00000000pepper-0.7.6+git44.996fc29/pyproject.toml[tool.pytest.ini_options]
addopts = "--showlocals --log-file /tmp/pepper-runtests.log --show-capture=no -ra"
testpaths = ["tests"]
norecursedirs = [".git", ".nox"]
usefixtures = ["pepperconfig"]
[tool.flake8]
max-line-length = 119
07070100000016000041ED00000000000000000000000267D9C77000000000000000000000000000000000000000000000002300000000pepper-0.7.6+git44.996fc29/scripts07070100000017000081A400000000000000000000000167D9C77000000228000000000000000000000000000000000000002A00000000pepper-0.7.6+git44.996fc29/scripts/pepper#!/usr/bin/env python
# Import Python Libraries
import logging
# Import Pepper Libraries
import pepper.script
try:
from logging import NullHandler
except ImportError: # Python < 2.7
class NullHandler(logging.Handler):
def emit(self, record): pass
logging.basicConfig(format='%(levelname)s %(asctime)s %(module)s: %(message)s')
logger = logging.getLogger('pepper')
logger.addHandler(NullHandler())
if __name__ == '__main__':
exit_code = pepper.script.Pepper()()
raise SystemExit(exit_code if exit_code is not None else 0)
07070100000018000081A400000000000000000000000167D9C77000000060000000000000000000000000000000000000002E00000000pepper-0.7.6+git44.996fc29/scripts/pepper.cmd@echo off
set _SCRIPTDIR=%~d0
set _SCRIPTPATH=%~p0
python "%_SCRIPTDIR%%_SCRIPTPATH%pepper" %*
07070100000019000081A400000000000000000000000167D9C770000005DD000000000000000000000000000000000000002400000000pepper-0.7.6+git44.996fc29/setup.py#!/usr/bin/env python
"""
A CLI front-end to a running salt-api system
"""
import setuptools
with open("README.rst") as fh:
long_description = fh.read()
setup_kwargs = {
"name": "salt-pepper",
"description": __doc__.strip(),
"author": "Seth House",
"author_email": "shouse@saltstack.com",
"url": "http://saltstack.com",
"long_description": long_description,
"long_description_content_type": "text/x-rst",
"use_scm_version": True,
"setup_requires": ["setuptools_scm"],
"classifiers": [
"Programming Language :: Python",
"Programming Language :: Cython",
"Programming Language :: Python :: 2.7",
"Programming Language :: Python :: 3.4",
"Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Development Status :: 5 - Production/Stable",
"Environment :: Console",
"Intended Audience :: Developers",
"Intended Audience :: Information Technology",
"Intended Audience :: System Administrators",
"License :: OSI Approved :: Apache Software License",
"Operating System :: POSIX :: Linux",
"Topic :: System :: Clustering",
"Topic :: System :: Distributed Computing",
],
"packages": [
"pepper",
],
"extras_require": {
"kerberos": ["requests-gssapi>=1.1.0"],
},
"scripts": [
"scripts/pepper",
],
}
if __name__ == "__main__":
setuptools.setup(**setup_kwargs)
0707010000001A000041ED00000000000000000000000267D9C77000000000000000000000000000000000000000000000002100000000pepper-0.7.6+git44.996fc29/tests0707010000001B000081A400000000000000000000000167D9C77000000000000000000000000000000000000000000000002D00000000pepper-0.7.6+git44.996fc29/tests/__init__.py0707010000001C000081A400000000000000000000000167D9C77000002306000000000000000000000000000000000000002D00000000pepper-0.7.6+git44.996fc29/tests/conftest.py# Import python libraries
import logging
import os.path
import shutil
import sys
import tempfile
import textwrap
import pytest
import salt.utils.yaml as yaml
from pytestskipmarkers.utils import ports
from saltfactories.utils import random_string
from saltfactories.utils import running_username
import pepper.script
# Import pytest libraries
# Import Salt Libraries
log = logging.getLogger(__name__)
@pytest.fixture(scope="session")
def sshd_config_dir(salt_factories):
config_dir = salt_factories.get_root_dir_for_daemon("sshd")
yield config_dir
shutil.rmtree(str(config_dir), ignore_errors=True)
@pytest.fixture(scope="session")
def session_sshd_server(salt_factories, sshd_config_dir, session_master):
sshd_config_dict = {
"Protocol": "2",
# Turn strict modes off so that we can operate in /tmp
"StrictModes": "no",
# Logging
"SyslogFacility": "AUTH",
"LogLevel": "INFO",
# Authentication:
"LoginGraceTime": "120",
"PermitRootLogin": "without-password",
"PubkeyAuthentication": "yes",
# Don't read the user's ~/.rhosts and ~/.shosts files
"IgnoreRhosts": "yes",
"HostbasedAuthentication": "no",
# To enable empty passwords, change to yes (NOT RECOMMENDED)
"PermitEmptyPasswords": "no",
# Change to yes to enable challenge-response passwords (beware issues with
# some PAM modules and threads)
"ChallengeResponseAuthentication": "no",
# Change to no to disable tunnelled clear text passwords
"PasswordAuthentication": "no",
"X11Forwarding": "no",
"X11DisplayOffset": "10",
"PrintMotd": "no",
"PrintLastLog": "yes",
"TCPKeepAlive": "yes",
"AcceptEnv": "LANG LC_*",
"UsePAM": "yes",
}
factory = salt_factories.get_sshd_daemon(
sshd_config_dict=sshd_config_dict,
config_dir=sshd_config_dir,
)
with factory.started():
yield factory
@pytest.fixture(scope="session")
def session_ssh_roster_config(session_sshd_server, session_master):
roster_contents = """
localhost:
host: 127.0.0.1
port: {}
user: {}
priv: {}
mine_functions:
test.arg: ['itworked']
""".format(
session_sshd_server.listen_port, running_username(), session_sshd_server.client_key
)
with pytest.helpers.temp_file(
"roster", roster_contents, session_master.config_dir
) as roster_file:
yield roster_file
@pytest.fixture(scope="session")
def salt_api_port():
"""
Returns an unused localhost port for the api port
"""
return ports.get_unused_localhost_port()
@pytest.fixture(scope="session")
def pepperconfig(salt_api_port):
config = textwrap.dedent(
"""
[main]
SALTAPI_URL=http://localhost:{0}/
SALTAPI_USER=pepper
SALTAPI_PASS=pepper
SALTAPI_EAUTH=sharedsecret
[pepper]
SALTAPI_URL=http://localhost:{0}/
SALTAPI_USER=pepper
SALTAPI_PASS=pepper
SALTAPI_EAUTH=sharedsecret
[baduser]
SALTAPI_URL=http://localhost:{0}/
SALTAPI_USER=saltdev
SALTAPI_PASS=saltdev
SALTAPI_EAUTH=pam
[badapi]
SALTAPI_URL=git://localhost:{0}/
[noapi]
SALTAPI_USER=pepper
SALTAPI_PASS=pepper
SALTAPI_EAUTH=sharedsecret
[noopts]
SALTAPI_URL=http://localhost:{0}/
""".format(
salt_api_port
)
)
with open("tests/.pepperrc", "w") as pepper_file:
print(config, file=pepper_file)
yield
os.remove("tests/.pepperrc")
@pytest.fixture
def pepper_client(session_salt_api, salt_api_port):
client = pepper.Pepper("http://localhost:{}".format(salt_api_port))
client.login("pepper", "pepper", "sharedsecret")
return client
@pytest.fixture
def tokfile():
tokdir = tempfile.mkdtemp()
yield os.path.join(tokdir, "peppertok.json")
shutil.rmtree(tokdir)
@pytest.fixture
def output_file():
"""
Returns the path to the salt master configuration file
"""
out_dir = tempfile.mkdtemp()
yield os.path.join(out_dir, "output")
shutil.rmtree(out_dir)
@pytest.fixture(params=["/run", "/login"])
def pepper_cli(request, session_salt_api, salt_api_port, output_file, session_sshd_server):
"""
Wrapper to invoke Pepper with common params and inside an empty env
"""
if request.config.getoption("--salt-api-backend") == "rest_tornado" and request.param == "/run":
pytest.xfail("rest_tornado does not support /run endpoint until next release")
def_args = [
"--out=json",
"--output-file={}".format(output_file),
"-c",
"tests/.pepperrc",
]
if request.param == "/run":
def_args = ["--run-uri"] + def_args
def _run_pepper_cli(*args, **kwargs):
sys.argv = ["pepper", "-p", kwargs.pop("profile", "main")] + def_args + list(args)
exitcode = pepper.script.Pepper()()
try:
with open(output_file) as result:
try:
return yaml.load(result)
except yaml.parser.ParserError:
result.seek(0)
return [
yaml.load("{0}}}".format(ret).strip('"'))
for ret in result.read().split('}"\n')
if ret
]
except Exception as exc:
log.error("ExitCode %s: %s", exitcode, exc)
return exitcode
return _run_pepper_cli
@pytest.fixture(scope="session")
def session_master_factory(request, salt_factories, session_master_config_overrides):
return salt_factories.salt_master_daemon(
random_string("master-"), overrides=session_master_config_overrides
)
@pytest.fixture(scope="session")
def session_master(session_master_factory):
with session_master_factory.started():
yield session_master_factory
@pytest.fixture(scope="session")
def session_master_config_overrides(request, salt_api_port, salt_api_backend):
return {
salt_api_backend: {
"port": salt_api_port,
"disable_ssl": True,
},
"external_auth": {
"sharedsecret": {
"pepper": [
".*",
"@jobs",
"@wheel",
"@runner",
],
},
},
"sharedsecret": "pepper",
"token_expire": 94670856,
"ignore_host_keys": True,
"ssh_wipe": True,
"netapi_enable_clients": [
"local",
"local_async",
"local_subset",
"ssh",
"runner",
"runner_async",
"wheel",
"wheel_async",
"run",
],
}
@pytest.helpers.register
def remove_stale_minion_key(master, minion_id):
"""Helper to remove a stale minion key."""
key_path = os.path.join(master.config["pki_dir"], "minions", minion_id)
if os.path.exists(key_path):
os.unlink(key_path)
else:
log.debug("The minion(id=%r) key was not found at %s", minion_id, key_path)
@pytest.fixture(scope="session")
def session_minion_factory(session_master_factory):
"""Return a factory for a randomly named minion connected to master."""
minion_factory = session_master_factory.salt_minion_daemon(random_string("minion-"))
minion_factory.after_terminate(
pytest.helpers.remove_stale_minion_key, session_master_factory, minion_factory.id
)
return minion_factory
@pytest.fixture(scope="session")
def session_minion(session_master, session_minion_factory): # noqa
assert session_master.is_running()
with session_minion_factory.started():
yield session_minion_factory
@pytest.fixture(scope="session")
def session_minion_id(session_minion):
return session_minion.id
@pytest.fixture(scope="session")
def salt_api_backend(request):
"""
Return the salt-api backend (cherrypy or tornado)
"""
backend = request.config.getoption("--salt-api-backend")
if backend is not None:
return backend
backend = request.config.getini("salt_api_backend")
if backend is not None:
return backend
return "rest_cherrypy"
@pytest.fixture(scope="session")
def session_salt_api_factory(session_master_factory):
return session_master_factory.salt_api_daemon()
@pytest.fixture(scope="session")
def session_salt_api(session_master, session_salt_api_factory):
assert session_master.is_running()
with session_salt_api_factory.started():
yield session_salt_api_factory
def pytest_addoption(parser):
parser.addoption(
"--salt-api-backend",
action="store",
default="rest_cherrypy",
help="which backend to use for salt-api, must be one of rest_cherrypy or rest_tornado",
)
0707010000001D000041ED00000000000000000000000267D9C77000000000000000000000000000000000000000000000002D00000000pepper-0.7.6+git44.996fc29/tests/integration0707010000001E000081A400000000000000000000000167D9C77000000000000000000000000000000000000000000000003900000000pepper-0.7.6+git44.996fc29/tests/integration/__init__.py0707010000001F000081A400000000000000000000000167D9C7700000089B000000000000000000000000000000000000003D00000000pepper-0.7.6+git44.996fc29/tests/integration/test_clients.pyimport json
import pathlib
import pytest
def test_local_bad_opts(pepper_cli):
with pytest.raises(SystemExit):
pepper_cli("*")
with pytest.raises(SystemExit):
pepper_cli("test.ping")
with pytest.raises(SystemExit):
pepper_cli("--client=ssh", "test.ping")
with pytest.raises(SystemExit):
pepper_cli("--client=ssh", "*")
@pytest.mark.xfail(
'config.getoption("--salt-api-backend") == "rest_tornado"',
reason="timeout kwarg isnt popped until next version of salt/tornado",
)
def test_runner_client(pepper_cli):
ret = pepper_cli(
"--timeout=123",
"--client=runner",
"test.arg",
"one",
"two=what",
"three={}".format(json.dumps({"hello": "world"})),
)
assert ret == {"args": ["one"], "kwargs": {"three": {"hello": "world"}, "two": "what"}}
@pytest.mark.xfail(
'config.getoption("--salt-api-backend") == "rest_tornado"',
reason="wheelClient unimplemented for now on tornado",
)
def test_wheel_client_arg(pepper_cli, session_minion):
ret = pepper_cli("--client=wheel", "minions.connected")
assert ret == [session_minion.id]
@pytest.mark.xfail(
'config.getoption("--salt-api-backend") == "rest_tornado"',
reason="wheelClient unimplemented for now on tornado",
)
def test_wheel_client_kwargs(pepper_cli, session_master):
ret = pepper_cli(
"--client=wheel",
"config.update_config",
"file_name=pepper",
"yaml_contents={}".format(json.dumps({"timeout": 5})),
)
assert ret == "Wrote pepper.conf"
default_include_dir = pathlib.Path(session_master.config["default_include"]).parent
pepper_config = pathlib.Path(session_master.config_dir) / default_include_dir / "pepper.conf"
assert pepper_config.exists
@pytest.mark.xfail(
'config.getoption("--salt-api-backend") == "rest_tornado"',
reason="sshClient unimplemented for now on tornado",
)
def test_ssh_client(pepper_cli, session_ssh_roster_config):
ret = pepper_cli("--client=ssh", "*", "test.ping")
assert ret["ssh"]["localhost"]["return"] is True
def test_bad_client(pepper_cli):
ret = pepper_cli("--client=whatever")
assert ret == 1
07070100000020000081A400000000000000000000000167D9C770000004B2000000000000000000000000000000000000003A00000000pepper-0.7.6+git44.996fc29/tests/integration/test_json.pyimport os
import shutil
import tempfile
def test_local_json(pepper_cli, session_minion_id):
json = '[{"tgt": "*", "fun": "test.ping", "client": "local"}]'
ret = pepper_cli("--json", json)
assert ret[session_minion_id] is True
def test_local_json_bad(pepper_cli):
json = "{what}"
ret = pepper_cli("--json", json)
assert ret == 1
def test_local_json_file(pepper_cli, session_minion_id):
tmpjson = os.path.join(tempfile.mkdtemp(), "json")
with open(tmpjson, "w") as tmpfile:
print(
'[{"client": "local", "tgt": "*", "fun": "test.ping"}]',
file=tmpfile,
)
ret = pepper_cli("--json-file", tmpjson)
shutil.rmtree(os.path.dirname(tmpjson))
assert ret[session_minion_id] is True
def test_local_json_file_bad(pepper_cli):
tmpjson = os.path.join(tempfile.mkdtemp(), "json")
with open(tmpjson, "w") as tmpfile:
print(
"{what}",
file=tmpfile,
)
ret = pepper_cli("--json-file", tmpjson)
shutil.rmtree(os.path.dirname(tmpjson))
assert ret == 1
def test_local_json_no_file(pepper_cli):
ret = pepper_cli("--json-file", "/tmp/wahteverfile")
assert ret == 1
07070100000021000081A400000000000000000000000167D9C7700000008B000000000000000000000000000000000000003B00000000pepper-0.7.6+git44.996fc29/tests/integration/test_local.pydef test_local(pepper_client, session_minion_id):
assert pepper_client.local("*", "test.ping")["return"][0][session_minion_id] is True
07070100000022000081A400000000000000000000000167D9C77000000612000000000000000000000000000000000000004000000000pepper-0.7.6+git44.996fc29/tests/integration/test_login_opts.pydef test_cli_opts(pepper_cli, session_minion_id, salt_api_port):
"""Test the using a profile"""
ret = pepper_cli(
"--saltapi-url=http://localhost:{}/".format(salt_api_port),
"--eauth=sharedsecret",
"--username=pepper",
"--password=pepper",
"*",
"test.ping",
profile="noprofile",
)
assert ret[session_minion_id] is True
def test_cli_opts_not_in_profile(pepper_cli, session_minion_id, salt_api_port):
"""Test the using a profile"""
ret = pepper_cli(
"--eauth=sharedsecret",
"--username=pepper",
"--password=pepper",
"*",
"test.ping",
profile="noopts",
)
assert ret[session_minion_id] is True
def test_cli_api_not_in_profile(pepper_cli, session_minion_id, salt_api_port):
"""Test the using a profile"""
ret = pepper_cli(
"--saltapi-url=http://localhost:{}/".format(salt_api_port),
"*",
"test.ping",
profile="noapi",
)
assert ret[session_minion_id] is True
def test_no_username(pepper_cli, session_minion_id, salt_api_port):
"""Test the using a profile"""
ret = pepper_cli(
"--non-interactive",
"*",
"test.ping",
profile="noopts",
)
assert ret == 1
def test_no_password(pepper_cli, session_minion_id, salt_api_port):
"""Test the using a profile"""
ret = pepper_cli(
"--username=pepper",
"--non-interactive",
"*",
"test.ping",
profile="noopts",
)
assert ret == 1
07070100000023000081A400000000000000000000000167D9C770000002DB000000000000000000000000000000000000003C00000000pepper-0.7.6+git44.996fc29/tests/integration/test_poller.pydef test_local_poll(pepper_cli, session_minion_id):
"""Test the returns poller for localclient"""
ret = pepper_cli("--fail-if-incomplete", "*", "test.sleep", "1")
assert ret[session_minion_id] is True
assert len(ret) == 1
def test_local_poll_long(pepper_cli, session_minion_id):
"""Test the returns poller for localclient"""
ret = pepper_cli("--fail-if-incomplete", "*", "test.sleep", "30")
assert ret[session_minion_id] is True
assert len(ret) == 1
def test_local_poll_timeout(pepper_cli, session_minion_id):
"""Test the returns poller for localclient"""
ret = pepper_cli("--timeout=5", "--fail-if-incomplete", "*", "test.sleep", "30")
assert ret == {"Failed": [session_minion_id]}
07070100000024000081A400000000000000000000000167D9C770000000BE000000000000000000000000000000000000003D00000000pepper-0.7.6+git44.996fc29/tests/integration/test_profile.pydef test_config_profile(pepper_cli, session_minion_id):
"""Test the using a profile"""
ret = pepper_cli("*", "test.ping", profile="pepper")
assert ret[session_minion_id] is True
07070100000025000081A400000000000000000000000167D9C7700000047D000000000000000000000000000000000000003B00000000pepper-0.7.6+git44.996fc29/tests/integration/test_token.pyimport json
import time
def test_local_token(tokfile, pepper_cli, session_minion_id):
"""Test local execution with token file"""
ret = pepper_cli("-x", tokfile, "--make-token", "*", "test.ping")
assert ret[session_minion_id] is True
def test_runner_token(tokfile, pepper_cli):
"""Test runner execution with token file"""
ret = pepper_cli("-x", tokfile, "--make-token", "--client", "runner", "test.metasyntactic")
exps = [
"foo",
"bar",
"baz",
"qux",
"quux",
"quuz",
"corge",
"grault",
"garply",
"waldo",
"fred",
"plugh",
"xyzzy",
"thud",
]
assert all(exp in ret for exp in exps)
def test_token_expire(tokfile, pepper_cli):
"""Test token override param"""
now = time.time()
pepper_cli("-x", tokfile, "--make-token", "--token-expire", "94670856", "*", "test.ping")
with open(tokfile) as tfile:
token = json.load(tfile)
diff = (now + float(94670856)) - token["expire"]
# Allow for 10-second window between request and master-side auth.
assert diff < 10
07070100000026000081A400000000000000000000000167D9C77000000255000000000000000000000000000000000000003D00000000pepper-0.7.6+git44.996fc29/tests/integration/test_vanilla.pyimport pytest
def test_local(pepper_cli, session_minion_id):
"""Sanity-check: Has at least one minion - /run - /login query type is parameterized"""
ret = pepper_cli("*", "test.ping")
assert ret[session_minion_id] is True
@pytest.mark.xfail(
'config.getoption("--salt-api-backend") == "rest_tornado"',
reason="this is broken in rest_tornado until future release",
)
def test_long_local(pepper_cli, session_minion_id):
"""Test a long call blocks until the return"""
ret = pepper_cli("--timeout=60", "*", "test.sleep", "30")
assert ret[session_minion_id] is True
07070100000027000081A400000000000000000000000167D9C77000000147000000000000000000000000000000000000003200000000pepper-0.7.6+git44.996fc29/tests/requirements.txtmock
pytest>=6.0.0
flake8-pyproject
pytest-rerunfailures
pytest-cov
pytest-salt-factories==1.0.4
CherryPy
setuptools_scm==7.1.0
importlib-metadata<5.0.0
pyzmq>=25.1.2 ; python_version < '3.13'
pyzmq>=26.2.0 ; python_version >= '3.13'
cryptography>=42.0.0; python_version < '3.13'
cryptography==42.0.2; python_version >= '3.13'
07070100000028000041ED00000000000000000000000267D9C77000000000000000000000000000000000000000000000002600000000pepper-0.7.6+git44.996fc29/tests/unit07070100000029000081A400000000000000000000000167D9C77000000000000000000000000000000000000000000000003200000000pepper-0.7.6+git44.996fc29/tests/unit/__init__.py0707010000002A000081A400000000000000000000000167D9C77000000271000000000000000000000000000000000000003300000000pepper-0.7.6+git44.996fc29/tests/unit/test_init.py# Import Python Libraries
import imp
import os
import shutil
import setuptools_scm
import pepper
# Import Pepper Libraries
def test_no_setup():
setuppath = os.path.join(os.path.dirname(pepper.__file__), os.pardir, "setup.py")
shutil.move(setuppath, setuppath + ".bak")
ptest = imp.load_source("ptest", os.path.join(os.path.dirname(pepper.__file__), "__init__.py"))
shutil.move(setuppath + ".bak", setuppath)
assert ptest.version == setuptools_scm.get_version()
assert ptest.sha is None
def test_version():
assert pepper.version == setuptools_scm.get_version()
assert pepper.sha is None
0707010000002B000081A400000000000000000000000167D9C77000000216000000000000000000000000000000000000003C00000000pepper-0.7.6+git44.996fc29/tests/unit/test_login_details.pyimport sys
from unittest.mock import MagicMock
from unittest.mock import patch
import pepper.cli
# Import Testing libraries
def test_interactive_logins():
sys.argv = ["pepper", "-c", "tests/.pepperrc", "-p", "noopts"]
with patch("pepper.cli.input", MagicMock(return_value="pepper")), patch(
"pepper.cli.getpass.getpass", MagicMock(return_value="pepper")
):
result = pepper.cli.PepperCli().get_login_details()
assert result["SALTAPI_USER"] == "pepper"
assert result["SALTAPI_PASS"] == "pepper"
0707010000002C000081A400000000000000000000000167D9C77000000123000000000000000000000000000000000000003D00000000pepper-0.7.6+git44.996fc29/tests/unit/test_retcodes_error.py# Import Python Libraries
import sys
import pytest
import pepper
# Import Pepper Libraries
# Import Testing Libraries
def test_fail_any():
sys.argv = ["pepper", "--fail-all", "--fail-any", "minion_id", "request"]
with pytest.raises(SystemExit):
pepper.script.Pepper()()
0707010000002D000081A400000000000000000000000167D9C7700000086D000000000000000000000000000000000000004100000000pepper-0.7.6+git44.996fc29/tests/unit/test_retcodes_fail_fail.py# Import Python Libraries
import sys
from unittest.mock import MagicMock
from unittest.mock import patch
import pepper
# Import Pepper Libraries
PAYLOAD = {
"return": [
{
"ezh.msk.ru": {
"jid": "20180414193904158892",
"ret": "/bin/sh: 123: command not found",
"retcode": 127,
},
"saltstack.ezh.msk.ru": {
"jid": "20180414193904158892",
"ret": "/bin/sh: 1: 123: not found",
"retcode": 127,
},
}
]
}
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_default():
sys.argv = ["pepper", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any():
sys.argv = ["pepper", "--fail-any", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any_none():
sys.argv = ["pepper", "--fail-any-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all():
sys.argv = ["pepper", "--fail-all", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all_none():
sys.argv = ["pepper", "--fail-all-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
0707010000002E000081A400000000000000000000000167D9C77000000847000000000000000000000000000000000000004100000000pepper-0.7.6+git44.996fc29/tests/unit/test_retcodes_fail_none.py# Import Python Libraries
import sys
from unittest.mock import MagicMock
from unittest.mock import patch
import pepper
# Import Pepper Libraries
PAYLOAD = {
"return": [
{
"saltstack.ezh.msk.ru": {
"jid": "20180414193904158892",
"ret": "/bin/sh: 123: command not found",
"retcode": 127,
},
"ezh.msk.ru": {
"jid": "20180414193904158892",
"ret": "Hello from SaltStack",
},
}
]
}
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_default():
sys.argv = ["pepper", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any():
sys.argv = ["pepper", "--fail-any", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any_none():
sys.argv = ["pepper", "--fail-any-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all():
sys.argv = ["pepper", "--fail-all", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all_none():
sys.argv = ["pepper", "--fail-all-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
0707010000002F000081A400000000000000000000000167D9C77000000812000000000000000000000000000000000000004100000000pepper-0.7.6+git44.996fc29/tests/unit/test_retcodes_fail_pass.py# Import Python Libraries
import sys
from unittest.mock import MagicMock
from unittest.mock import patch
import pepper
# Import Pepper Libraries
PAYLOAD = {
"return": [
{
"ezh.msk.ru": {
"jid": "20180414193904158892",
"ret": "/bin/sh: 123: command not found",
"retcode": 127,
},
"saltstack.ezh.msk.ru": {"jid": "20180414193904158892", "ret": "pass", "retcode": 0},
}
]
}
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_default():
sys.argv = ["pepper", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any():
sys.argv = ["pepper", "--fail-any", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any_none():
sys.argv = ["pepper", "--fail-any-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all():
sys.argv = ["pepper", "--fail-all", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all_none():
sys.argv = ["pepper", "--fail-all-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
07070100000030000081A400000000000000000000000167D9C77000000816000000000000000000000000000000000000004100000000pepper-0.7.6+git44.996fc29/tests/unit/test_retcodes_none_none.py# Import Python Libraries
import sys
from unittest.mock import MagicMock
from unittest.mock import patch
import pepper
# Import Pepper Libraries
PAYLOAD = {
"return": [
{
"saltstack.ezh.msk.ru": {
"jid": "20180414193904158892",
"ret": "Hello from SaltStack",
},
"ezh.msk.ru": {
"jid": "20180414193904158892",
"ret": "Hello from SaltStack",
},
}
]
}
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_default():
sys.argv = ["pepper", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any():
sys.argv = ["pepper", "--fail-any", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any_none():
sys.argv = ["pepper", "--fail-any-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == -1
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all():
sys.argv = ["pepper", "--fail-all", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all_none():
sys.argv = ["pepper", "--fail-all-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == -1
07070100000031000081A400000000000000000000000167D9C77000000812000000000000000000000000000000000000004100000000pepper-0.7.6+git44.996fc29/tests/unit/test_retcodes_pass_fail.py# Import Python Libraries
import sys
from unittest.mock import MagicMock
from unittest.mock import patch
import pepper
# Import Pepper Libraries
PAYLOAD = {
"return": [
{
"saltstack.ezh.msk.ru": {"jid": "20180414193904158892", "ret": "pass", "retcode": 0},
"ezh.msk.ru": {
"jid": "20180414193904158892",
"ret": "/bin/sh: 123: command not found",
"retcode": 127,
},
}
]
}
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_default():
sys.argv = ["pepper", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any():
sys.argv = ["pepper", "--fail-any", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any_none():
sys.argv = ["pepper", "--fail-any-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 127
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all():
sys.argv = ["pepper", "--fail-all", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all_none():
sys.argv = ["pepper", "--fail-all-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
07070100000032000081A400000000000000000000000167D9C770000007E3000000000000000000000000000000000000004100000000pepper-0.7.6+git44.996fc29/tests/unit/test_retcodes_pass_none.py# Import Python Libraries
import sys
from unittest.mock import MagicMock
from unittest.mock import patch
import pepper
# Import Pepper Libraries
PAYLOAD = {
"return": [
{
"saltstack.ezh.msk.ru": {"jid": "20180414193904158892", "ret": "pass", "retcode": 0},
"ezh.msk.ru": {
"jid": "20180414193904158892",
"ret": "Hello from SaltStack",
},
}
]
}
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_default():
sys.argv = ["pepper", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any():
sys.argv = ["pepper", "--fail-any", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any_none():
sys.argv = ["pepper", "--fail-any-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all():
sys.argv = ["pepper", "--fail-all", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all_none():
sys.argv = ["pepper", "--fail-all-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
07070100000033000081A400000000000000000000000167D9C770000007B2000000000000000000000000000000000000004100000000pepper-0.7.6+git44.996fc29/tests/unit/test_retcodes_pass_pass.py# Import Python Libraries
import sys
from unittest.mock import MagicMock
from unittest.mock import patch
import pepper
# Import Pepper Libraries
PAYLOAD = {
"return": [
{
"ezh.msk.ru": {"jid": "20180414193904158892", "ret": "pass", "retcode": 0},
"saltstack.ezh.msk.ru": {"jid": "20180414193904158892", "ret": "pass", "retcode": 0},
}
]
}
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_default():
sys.argv = ["pepper", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any():
sys.argv = ["pepper", "--fail-any", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_any_none():
sys.argv = ["pepper", "--fail-any-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all():
sys.argv = ["pepper", "--fail-all", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
@patch("pepper.cli.PepperCli.login", MagicMock(side_effect=lambda arg: None))
@patch("pepper.cli.PepperCli.low", MagicMock(side_effect=lambda api, load: PAYLOAD))
def test_fail_all_none():
sys.argv = ["pepper", "--fail-all-none", "minion_id", "request"]
ret_code = pepper.script.Pepper()()
assert ret_code == 0
07070100000034000081A400000000000000000000000167D9C770000005B7000000000000000000000000000000000000003400000000pepper-0.7.6+git44.996fc29/tests/unit/test_token.pyimport json
import sys
from unittest.mock import MagicMock
from unittest.mock import mock_open
from unittest.mock import patch
import pepper.cli
# Import Pepper Libraries
# Import Testing Libraries
def test_token():
sys.argv = ["pepper", "*", "test.ping"]
client = pepper.cli.PepperCli()
client.options.mktoken = True
mock_data = (
'{"perms": [".*", "@runner", "@wheel", "@jobs"], "start": 1529967752.516165, '
'"token": "7130faa1e17f935d5f2702465cafdc73212d64d0", "expire": 1529968905.1131861, '
'"user": "pepper", "eauth": "pam"}\n'
)
mock_api = MagicMock()
mock_api.login = MagicMock(return_value=mock_data)
with patch("pepper.cli.open", mock_open(read_data=mock_data)), patch(
"pepper.cli.PepperCli.get_login_details", MagicMock(return_value=mock_data)
), patch("pepper.cli.PepperCli.parse_login", MagicMock(return_value={})), patch(
"os.remove", MagicMock(return_value=None)
), patch(
"json.dump", MagicMock(side_effect=Exception("Test Error"))
):
ret1 = client.login(mock_api)
with patch("os.path.isfile", MagicMock(return_value=False)):
ret2 = client.login(mock_api)
with patch("time.time", MagicMock(return_value=1529968044.133632)):
ret3 = client.login(mock_api)
assert json.loads(ret1) == json.loads(mock_data)
assert json.loads(ret2) == json.loads(mock_data)
assert ret3 == json.loads(mock_data)
07070100000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000B00000000TRAILER!!!224 blocks
