mirror of
https://github.com/bvanroll/python-vault-db.git
synced 2025-08-29 20:12:43 +00:00
Compare commits
5 Commits
Author | SHA1 | Date | |
---|---|---|---|
9c4581cb72 | |||
fbf0277dd1 | |||
493d13f779 | |||
577d112b89 | |||
3cf1fa2dbd |
3
.github/workflows/python-publish.yml
vendored
3
.github/workflows/python-publish.yml
vendored
@@ -22,6 +22,9 @@ jobs:
|
|||||||
run: |
|
run: |
|
||||||
python -m pip install --upgrade pip
|
python -m pip install --upgrade pip
|
||||||
pip install setuptools wheel twine
|
pip install setuptools wheel twine
|
||||||
|
- name: Test #TODO use a testing solution that won't be deprecated soon
|
||||||
|
run: |
|
||||||
|
python setup.py test
|
||||||
- name: Build and publish
|
- name: Build and publish
|
||||||
env:
|
env:
|
||||||
TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }}
|
TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }}
|
||||||
|
21
.github/workflows/python-test.yml
vendored
Normal file
21
.github/workflows/python-test.yml
vendored
Normal file
@@ -0,0 +1,21 @@
|
|||||||
|
name: Unit test runner
|
||||||
|
|
||||||
|
on: push
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
test:
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v2
|
||||||
|
- name: Set up Python
|
||||||
|
uses: actions/setup-python@v2
|
||||||
|
with:
|
||||||
|
python-version: '3.x'
|
||||||
|
- name: Install dependencies
|
||||||
|
run: |
|
||||||
|
python -m pip install --upgrade pip
|
||||||
|
pip install setuptools wheel twine
|
||||||
|
- name: Test #TODO use a testing solution that won't be deprecated soon
|
||||||
|
run: |
|
||||||
|
python setup.py test
|
||||||
|
|
7
setup.py
7
setup.py
@@ -4,9 +4,10 @@ with open("README.md", "r", encoding="utf-8") as fh:
|
|||||||
long_description = fh.read()
|
long_description = fh.read()
|
||||||
|
|
||||||
setuptools.setup(
|
setuptools.setup(
|
||||||
name="VaultDb",
|
name="vaultdb",
|
||||||
version="0.0.1",
|
version="0.0.1",
|
||||||
version_config=True,
|
version_config=True,
|
||||||
|
test_suite="test",
|
||||||
setup_requires=['setuptools-git-versioning'],
|
setup_requires=['setuptools-git-versioning'],
|
||||||
author="Beppe Vanrolleghem",
|
author="Beppe Vanrolleghem",
|
||||||
author_email="beppe.vanrolleghem@gmail.com",
|
author_email="beppe.vanrolleghem@gmail.com",
|
||||||
@@ -22,7 +23,7 @@ setuptools.setup(
|
|||||||
"License :: OSI Approved :: MIT License",
|
"License :: OSI Approved :: MIT License",
|
||||||
"Operating System :: OS Independent",
|
"Operating System :: OS Independent",
|
||||||
],
|
],
|
||||||
package_dir={"": "VaultDb"},
|
package_dir={"": "src"},
|
||||||
packages=setuptools.find_packages(where="VaultDb"),
|
packages=setuptools.find_packages(where="src"),
|
||||||
python_requires=">=3.6",
|
python_requires=">=3.6",
|
||||||
)
|
)
|
||||||
|
@@ -1,48 +0,0 @@
|
|||||||
import requests as rq
|
|
||||||
import datetime as dt
|
|
||||||
|
|
||||||
class Database2:
|
|
||||||
def __init__(self, dbname, vault_url, token, vault_port=8200):
|
|
||||||
self.dbname = dbname
|
|
||||||
self.url = vault_url + ":" + str(vault_port) + "/v1/database/creds/" + dbname
|
|
||||||
self.lastReq = None #TODO some datetime in here
|
|
||||||
self.ttl = None
|
|
||||||
self.token = token
|
|
||||||
self.username = ""
|
|
||||||
self.password = ""
|
|
||||||
self.get_creds()
|
|
||||||
|
|
||||||
def check_valid(self):
|
|
||||||
#TODO check datetime for current datetime
|
|
||||||
if self.ttl == None:
|
|
||||||
return False
|
|
||||||
return True
|
|
||||||
|
|
||||||
def update_creds(self):
|
|
||||||
r = rq.get(url=self.url, headers={"X-Vault-Token": self.token, "Content-Type": "application/json"})
|
|
||||||
# print(r.status_code)
|
|
||||||
# print(r.text)
|
|
||||||
if r.status_code != 200:
|
|
||||||
raise Exception("status code was nog 200")
|
|
||||||
data = r.json()
|
|
||||||
# TODO set datetime for current datetime
|
|
||||||
# TODO set these to the correct response values
|
|
||||||
self.username = data["data"]["username"]
|
|
||||||
self.password = data["data"]["password"]
|
|
||||||
self.lease_id = data["lease_id"]
|
|
||||||
self.request_id = data["request_id"]
|
|
||||||
self.wrap_info = data["wrap_info"]
|
|
||||||
self.warnings = data["warnings"]
|
|
||||||
self.auth = data["auth"]
|
|
||||||
self.ttl = dt.datetime.now() + dt.timedelta(seconds=float(data["lease_duration"]))
|
|
||||||
|
|
||||||
|
|
||||||
def get_creds(self):
|
|
||||||
if not (self.check_valid()):
|
|
||||||
self.update_creds()
|
|
||||||
return {"username": self.username, "password": self.password}
|
|
||||||
|
|
||||||
def get_username(self):
|
|
||||||
if not (self.check_valid()):
|
|
||||||
self.update_creds()
|
|
||||||
return self.username
|
|
7
test/__init__.py
Normal file
7
test/__init__.py
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
import unittest
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
@@ -4,8 +4,17 @@ import os
|
|||||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
||||||
import src
|
import src
|
||||||
|
|
||||||
class MyTestCase(unittest.TestCase):
|
class DatabaseClassTests(unittest.TestCase):
|
||||||
def test_psql(self):
|
|
||||||
|
def test_vault_connection_error(self):
|
||||||
|
#TODO give some "valid" url
|
||||||
|
self.fail()
|
||||||
|
|
||||||
|
def test_vault_non_existent_database(self):
|
||||||
|
#TODO test non existent_database
|
||||||
|
self.fail()
|
||||||
|
|
||||||
|
def test_psql_connection(self):
|
||||||
#TODO exec psql connect
|
#TODO exec psql connect
|
||||||
print("hi")
|
print("hi")
|
||||||
test = src.Database()
|
test = src.Database()
|
||||||
@@ -13,18 +22,24 @@ class MyTestCase(unittest.TestCase):
|
|||||||
self.assertEqual(test.valid, False)
|
self.assertEqual(test.valid, False)
|
||||||
|
|
||||||
def test_psql_invalid_vault_url(self):
|
def test_psql_invalid_vault_url(self):
|
||||||
#TODO figure out how to fail unit test check
|
|
||||||
try:
|
try:
|
||||||
src.Database(dbname="sdfk", vault_url="localhost", vault_port=8200, token="bla")
|
src.Database(dbname="sdfk", vault_url="localhost", vault_port=8200, token="bla")
|
||||||
except:
|
except:
|
||||||
print("did it")
|
#TODO check the err msg
|
||||||
return
|
return
|
||||||
|
else:
|
||||||
|
self.fail()
|
||||||
|
|
||||||
|
|
||||||
def test_psql_invalid_vault_port(self):
|
def test_psql_invalid_vault_port(self):
|
||||||
try:
|
try:
|
||||||
src.Database(dbname="sdfk", vault_url="localhost", vault_port="a", token="bla")
|
src.Database(dbname="sdfk", vault_url="localhost", vault_port="a", token="bla")
|
||||||
except:
|
except:
|
||||||
|
#TODO check the error message
|
||||||
return
|
return
|
||||||
|
else:
|
||||||
|
self.fail()
|
||||||
|
|
||||||
|
|
||||||
def test_psql_valid(self):
|
def test_psql_valid(self):
|
||||||
print(src.Database(dbname="psql", vault_url="http://localhost", token="s.GoR2nisHPeKU1vOaw9hZ5L7h").get_creds())
|
print(src.Database(dbname="psql", vault_url="http://localhost", token="s.GoR2nisHPeKU1vOaw9hZ5L7h").get_creds())
|
Reference in New Issue
Block a user