Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • perun/perun-proxyidp/perun-proxy-utils
1 result
Show changes
Commits on Source (5)
# [1.2.0](https://gitlab.ics.muni.cz/perun-proxy-aai/python/perun-proxy-utils/compare/v1.1.0...v1.2.0) (2023-06-06)
### Features
* script calling non-python monitoring scripts ([159b00b](https://gitlab.ics.muni.cz/perun-proxy-aai/python/perun-proxy-utils/commit/159b00b5e0707306a1e1442d4dcdf3018587dcbc))
* script to check RPC availability ([ae1c2d2](https://gitlab.ics.muni.cz/perun-proxy-aai/python/perun-proxy-utils/commit/ae1c2d24add1804a911153d8efc7db2e9f8e8ea3))
# [1.1.0](https://gitlab.ics.muni.cz/perun-proxy-aai/python/perun-proxy-utils/compare/v1.0.0...v1.1.0) (2023-04-11)
......
#!/usr/bin/env python3
import argparse
import subprocess
import sys
"""
general script to run non-python checks by a custom-defined command
"""
def get_args():
"""
Supports the command-line arguments listed below.
"""
parser = argparse.ArgumentParser(description="Custom command to run")
parser.add_argument(
"-c",
"--command",
required=True,
help="whole command to be executed",
)
return parser.parse_args()
def main():
args = get_args()
result = subprocess.run(args.command, shell=True, text=True, capture_output=True)
print(result.stdout, end="")
return result.returncode
if __name__ == "__main__":
sys.exit(main())
#!/usr/bin/env python3
import argparse
import re
import sys
import time
import requests
"""
check RPC API is available
"""
def get_args():
"""
Supports the command-line arguments listed below.
"""
parser = argparse.ArgumentParser(description="Check RPC status")
parser.add_argument(
"-u",
"--username",
required=True,
help="username for IdP",
)
parser.add_argument(
"-p",
"--password",
required=True,
help="password for IdP",
)
parser.add_argument(
"-d",
"--domain",
help="RPC domain with authentication method (e.g. 'perun.cesnet.cz/ba')",
required=True,
)
parser.add_argument(
"-i",
"--id",
type=int,
help="valid userId - This id will be used in getUserById call",
required=True,
)
return parser.parse_args()
def call_api(auth, url, user_id):
start_time = time.time()
try:
response = requests.get(url, timeout=10, auth=auth)
rpc_result = response.text
except requests.Timeout:
rpc_result = "Request timeout"
end_time = time.time()
total_time = end_time - start_time
if re.search(r'"id":' + str(user_id), rpc_result):
print(f"0 check_rpc_status - total_time={total_time:.4f} OK")
return 0
else:
rpc_result = rpc_result.replace("\n", " ")
print(f"2 check_rpc_status - total_time={total_time:.4f} {rpc_result}")
return 2
def main():
args = get_args()
auth = (args.username, args.password)
url = f"https://{args.domain}/rpc/json/usersManager/getUserById?id={args.id}"
return call_api(auth, url, args.id)
if __name__ == "__main__":
sys.exit(main())
[metadata]
version = 1.1.0
version = 1.2.0
license_files = LICENSE
......@@ -13,5 +13,6 @@ setuptools.setup(
"asyncssh~=2.13",
"docker~=6.0",
"beautifulsoup4~=4.12",
"requests~=2.31",
],
)