config: Add Config.rt_credentials method.
This loads settings from the same environment variables and ~/.rtrc file as the rt CLI. Note that it does *not* support RTCONFIG and the config file searching, because right now that seems like more work for more trouble to me.
This commit is contained in:
parent
5140ca64f6
commit
8d3816a8fd
5 changed files with 138 additions and 7 deletions
|
@ -18,12 +18,58 @@ import os
|
|||
|
||||
from pathlib import Path
|
||||
from typing import (
|
||||
NamedTuple,
|
||||
Optional,
|
||||
)
|
||||
|
||||
class RTCredentials(NamedTuple):
|
||||
server: Optional[str] = None
|
||||
user: Optional[str] = None
|
||||
passwd: Optional[str] = None
|
||||
auth: Optional[str] = None
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> 'RTCredentials':
|
||||
values = dict(cls._field_defaults)
|
||||
for key in values:
|
||||
env_key = 'RT{}'.format(key.upper())
|
||||
try:
|
||||
values[key] = os.environ[env_key]
|
||||
except KeyError:
|
||||
pass
|
||||
return cls(**values)
|
||||
|
||||
@classmethod
|
||||
def from_rtrc(cls) -> 'RTCredentials':
|
||||
values = dict(cls._field_defaults)
|
||||
rtrc_path = Path.home() / '.rtrc'
|
||||
try:
|
||||
with rtrc_path.open() as rtrc_file:
|
||||
for line in rtrc_file:
|
||||
try:
|
||||
key, value = line.split(None, 1)
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
if key in values:
|
||||
values[key] = value.rstrip('\n')
|
||||
except OSError:
|
||||
return cls()
|
||||
else:
|
||||
return cls(**values)
|
||||
|
||||
|
||||
class Config:
|
||||
def repository_path(self) -> Optional[Path]:
|
||||
try:
|
||||
return Path(os.environ['CONSERVANCY_REPOSITORY'])
|
||||
except (KeyError, ValueError):
|
||||
return None
|
||||
|
||||
def rt_credentials(self) -> RTCredentials:
|
||||
all_creds = zip(
|
||||
RTCredentials.from_env(),
|
||||
RTCredentials.from_rtrc(),
|
||||
RTCredentials(auth='rt'),
|
||||
)
|
||||
return RTCredentials._make(v0 or v1 or v2 for v0, v1, v2 in all_creds)
|
||||
|
|
|
@ -7,3 +7,8 @@ from . import testutil
|
|||
@pytest.fixture(scope='session', autouse=True)
|
||||
def clean_environment():
|
||||
os.environ.pop('CONSERVANCY_REPOSITORY', None)
|
||||
os.environ.pop('RTAUTH', None)
|
||||
os.environ.pop('RTPASSWD', None)
|
||||
os.environ.pop('RTSERVER', None)
|
||||
os.environ.pop('RTUSER', None)
|
||||
os.environ['HOME'] = str(testutil.test_path('userconfig'))
|
||||
|
|
|
@ -17,8 +17,37 @@
|
|||
import contextlib
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from . import testutil
|
||||
|
||||
from conservancy_beancount import config as config_mod
|
||||
|
||||
RT_ENV_KEYS = (
|
||||
'RTSERVER',
|
||||
'RTUSER',
|
||||
'RTPASSWD',
|
||||
'RTAUTH',
|
||||
)
|
||||
|
||||
RT_ENV_CREDS = (
|
||||
'https://example.org/envrt',
|
||||
'envuser',
|
||||
'env password',
|
||||
'gssapi',
|
||||
)
|
||||
|
||||
RT_FILE_CREDS = (
|
||||
'https://example.org/filert',
|
||||
'fileuser',
|
||||
'file password',
|
||||
'basic',
|
||||
)
|
||||
|
||||
@pytest.fixture
|
||||
def rt_environ():
|
||||
return dict(zip(RT_ENV_KEYS, RT_ENV_CREDS))
|
||||
|
||||
def _update_environ(updates):
|
||||
for key, value in updates.items():
|
||||
if value is None:
|
||||
|
@ -41,3 +70,42 @@ def test_repository_from_environment():
|
|||
def test_no_repository():
|
||||
config = config_mod.Config()
|
||||
assert config.repository_path() is None
|
||||
|
||||
def test_no_rt_credentials():
|
||||
with update_environ(HOME=testutil.TESTS_DIR):
|
||||
config = config_mod.Config()
|
||||
rt_credentials = config.rt_credentials()
|
||||
assert rt_credentials.server is None
|
||||
assert rt_credentials.user is None
|
||||
assert rt_credentials.passwd is None
|
||||
assert rt_credentials.auth == 'rt'
|
||||
|
||||
def test_rt_credentials_from_file():
|
||||
config = config_mod.Config()
|
||||
rt_credentials = config.rt_credentials()
|
||||
assert rt_credentials == RT_FILE_CREDS
|
||||
|
||||
def test_rt_credentials_from_environment(rt_environ):
|
||||
with update_environ(**rt_environ):
|
||||
config = config_mod.Config()
|
||||
rt_credentials = config.rt_credentials()
|
||||
assert rt_credentials == RT_ENV_CREDS
|
||||
|
||||
@pytest.mark.parametrize('index,drop_key', enumerate(RT_ENV_KEYS))
|
||||
def test_rt_credentials_from_file_and_environment_mixed(rt_environ, index, drop_key):
|
||||
del rt_environ[drop_key]
|
||||
with update_environ(**rt_environ):
|
||||
config = config_mod.Config()
|
||||
rt_credentials = config.rt_credentials()
|
||||
expected = list(RT_ENV_CREDS)
|
||||
expected[index] = RT_FILE_CREDS[index]
|
||||
assert rt_credentials == tuple(expected)
|
||||
|
||||
def test_rt_credentials_from_all_sources_mixed(tmp_path):
|
||||
server = 'https://example.org/mixedrt'
|
||||
with (tmp_path / '.rtrc').open('w') as rtrc_file:
|
||||
print('user basemix', 'passwd mixed up', file=rtrc_file, sep='\n')
|
||||
with update_environ(HOME=tmp_path, RTSERVER=server, RTUSER='mixedup'):
|
||||
config = config_mod.Config()
|
||||
rt_credentials = config.rt_credentials()
|
||||
assert rt_credentials == (server, 'mixedup', 'mixed up', 'rt')
|
||||
|
|
|
@ -27,6 +27,7 @@ FUTURE_DATE = datetime.date.today() + datetime.timedelta(days=365 * 99)
|
|||
FY_START_DATE = datetime.date(2020, 3, 1)
|
||||
FY_MID_DATE = datetime.date(2020, 9, 1)
|
||||
PAST_DATE = datetime.date(2000, 1, 1)
|
||||
TESTS_DIR = Path(__file__).parent
|
||||
|
||||
def check_post_meta(txn, *expected_meta, default=None):
|
||||
assert len(txn.postings) == len(expected_meta)
|
||||
|
@ -42,6 +43,14 @@ def check_post_meta(txn, *expected_meta, default=None):
|
|||
def parse_date(s, fmt='%Y-%m-%d'):
|
||||
return datetime.datetime.strptime(s, fmt).date()
|
||||
|
||||
def test_path(s):
|
||||
if s is None:
|
||||
return s
|
||||
s = Path(s)
|
||||
if not s.is_absolute():
|
||||
s = TESTS_DIR / s
|
||||
return s
|
||||
|
||||
def Posting(account, number,
|
||||
currency='USD', cost=None, price=None, flag=None,
|
||||
**meta):
|
||||
|
@ -98,14 +107,8 @@ class Transaction:
|
|||
|
||||
|
||||
class TestConfig:
|
||||
TESTS_DIR = Path(__file__).parent
|
||||
|
||||
def __init__(self, repo_path=None):
|
||||
if repo_path is not None:
|
||||
repo_path = Path(repo_path)
|
||||
if not repo_path.is_absolute():
|
||||
repo_path = Path(self.TESTS_DIR, repo_path)
|
||||
self.repo_path = repo_path
|
||||
self.repo_path = test_path(repo_path)
|
||||
|
||||
def repository_path(self):
|
||||
return self.repo_path
|
||||
|
|
9
tests/userconfig/.rtrc
Normal file
9
tests/userconfig/.rtrc
Normal file
|
@ -0,0 +1,9 @@
|
|||
server https://example.org/filert
|
||||
|
||||
user fileuser
|
||||
|
||||
# Value tests that spaces are handled correctly.
|
||||
passwd file password
|
||||
|
||||
# Need to handle auth too!
|
||||
auth basic
|
Loading…
Reference in a new issue