rtutil: Add RT.txn_with_urls() method.
This commit is contained in:
parent
701ccdc192
commit
999ca2c5e1
2 changed files with 92 additions and 0 deletions
|
@ -25,8 +25,13 @@ import rt
|
||||||
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
from . import data
|
||||||
|
from beancount.core import data as bc_data
|
||||||
|
|
||||||
from typing import (
|
from typing import (
|
||||||
|
overload,
|
||||||
Callable,
|
Callable,
|
||||||
|
Iterable,
|
||||||
Iterator,
|
Iterator,
|
||||||
MutableMapping,
|
MutableMapping,
|
||||||
Optional,
|
Optional,
|
||||||
|
@ -34,6 +39,9 @@ from typing import (
|
||||||
Tuple,
|
Tuple,
|
||||||
Union,
|
Union,
|
||||||
)
|
)
|
||||||
|
from .beancount_types import (
|
||||||
|
Transaction,
|
||||||
|
)
|
||||||
|
|
||||||
RTId = Union[int, str]
|
RTId = Union[int, str]
|
||||||
TicketAttachmentIds = Tuple[str, Optional[str]]
|
TicketAttachmentIds = Tuple[str, Optional[str]]
|
||||||
|
@ -263,6 +271,53 @@ class RT:
|
||||||
)
|
)
|
||||||
return self._extend_url(path_tail)
|
return self._extend_url(path_tail)
|
||||||
|
|
||||||
|
def _urls(self, links: Iterable[str]) -> Iterator[str]:
|
||||||
|
for link in links:
|
||||||
|
parsed = self.parse(link)
|
||||||
|
if parsed is None:
|
||||||
|
yield link
|
||||||
|
else:
|
||||||
|
ticket_id, attachment_id = parsed
|
||||||
|
url = self.url(ticket_id, attachment_id)
|
||||||
|
yield f'<{url}>'
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def _meta_with_urls(self, meta: None) -> None: ...
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def _meta_with_urls(self, meta: bc_data.Meta) -> bc_data.Meta: ...
|
||||||
|
|
||||||
|
def _meta_with_urls(self, meta: Optional[bc_data.Meta]) -> Optional[bc_data.Meta]:
|
||||||
|
if meta is None:
|
||||||
|
return None
|
||||||
|
link_meta = data.Metadata(meta)
|
||||||
|
retval = meta.copy()
|
||||||
|
for key in data.LINK_METADATA:
|
||||||
|
try:
|
||||||
|
links = link_meta.get_links(key)
|
||||||
|
except TypeError:
|
||||||
|
continue
|
||||||
|
if links:
|
||||||
|
retval[key] = ' '.join(self._urls(links))
|
||||||
|
return retval
|
||||||
|
|
||||||
|
def txn_with_urls(self, txn: Transaction) -> Transaction:
|
||||||
|
"""Copy a transaction with RT references replaced with web URLs
|
||||||
|
|
||||||
|
Given a Beancount Transaction, this method returns a Transaction
|
||||||
|
that's identical, except any references to RT in the metadata for
|
||||||
|
the Transaction and its Postings are replaced with web URLs.
|
||||||
|
This is useful for reporting tools that want to format the
|
||||||
|
transaction with URLs that are recognizable by other tools.
|
||||||
|
"""
|
||||||
|
# mypy doesn't recognize that postings is a valid argument, probably a
|
||||||
|
# bug in the NamedTuple→Directive→Transaction hierarchy.
|
||||||
|
return txn._replace( # type:ignore[call-arg]
|
||||||
|
meta=self._meta_with_urls(txn.meta),
|
||||||
|
postings=[post._replace(meta=self._meta_with_urls(post.meta))
|
||||||
|
for post in txn.postings],
|
||||||
|
)
|
||||||
|
|
||||||
def exists(self, ticket_id: RTId, attachment_id: Optional[RTId]=None) -> bool:
|
def exists(self, ticket_id: RTId, attachment_id: Optional[RTId]=None) -> bool:
|
||||||
return self.url(ticket_id, attachment_id) is not None
|
return self.url(ticket_id, attachment_id) is not None
|
||||||
|
|
||||||
|
|
|
@ -39,6 +39,11 @@ EXPECTED_URLS = [
|
||||||
(9, None, None),
|
(9, None, None),
|
||||||
]
|
]
|
||||||
|
|
||||||
|
EXPECTED_URLS_MAP = {
|
||||||
|
(ticket_id, attachment_id): url
|
||||||
|
for ticket_id, attachment_id, url in EXPECTED_URLS
|
||||||
|
}
|
||||||
|
|
||||||
@pytest.fixture(scope='module')
|
@pytest.fixture(scope='module')
|
||||||
def rt():
|
def rt():
|
||||||
client = testutil.RTClient()
|
client = testutil.RTClient()
|
||||||
|
@ -203,3 +208,35 @@ def test_results_not_found_only_in_transient_cache(new_client):
|
||||||
new_client.TICKET_DATA['9'] = [('99', '(Unnamed)', 'text/plain', '0b')]
|
new_client.TICKET_DATA['9'] = [('99', '(Unnamed)', 'text/plain', '0b')]
|
||||||
assert not rt1.exists(9)
|
assert not rt1.exists(9)
|
||||||
assert rt2.exists(9)
|
assert rt2.exists(9)
|
||||||
|
|
||||||
|
def test_txn_with_urls(new_client):
|
||||||
|
txn_meta = {
|
||||||
|
'rt-id': 'rt:1',
|
||||||
|
'contract': 'RepoLink.pdf',
|
||||||
|
'statement': 'doc1.txt rt:1/4 doc2.txt',
|
||||||
|
}
|
||||||
|
txn = testutil.Transaction(**txn_meta, postings=[
|
||||||
|
('Income:Donations', -10, {'receipt': 'rt:2/13 donation.txt'}),
|
||||||
|
('Assets:Cash', 10, {'receipt': 'cash.png rt:2/14'}),
|
||||||
|
])
|
||||||
|
rt = rtutil.RT(new_client)
|
||||||
|
actual = rt.txn_with_urls(txn)
|
||||||
|
def check(source, key, ticket_id, attachment_id=None):
|
||||||
|
url_path = EXPECTED_URLS_MAP[(ticket_id, attachment_id)]
|
||||||
|
assert f'<{DEFAULT_RT_URL}{url_path}>' in source.meta[key]
|
||||||
|
expected_keys = set(txn_meta)
|
||||||
|
expected_keys.update(['filename', 'lineno'])
|
||||||
|
assert set(actual.meta) == expected_keys
|
||||||
|
check(actual, 'rt-id', 1)
|
||||||
|
assert actual.meta['contract'] == txn_meta['contract']
|
||||||
|
assert actual.meta['statement'].startswith('doc1.txt ')
|
||||||
|
check(actual, 'statement', 1, 4)
|
||||||
|
check(actual.postings[0], 'receipt', 2, 13)
|
||||||
|
assert actual.postings[0].meta['receipt'].endswith(' donation.txt')
|
||||||
|
check(actual.postings[1], 'receipt', 2, 14)
|
||||||
|
assert actual.postings[1].meta['receipt'].startswith('cash.png ')
|
||||||
|
# Check the original transaction is unchanged
|
||||||
|
for key, expected in txn_meta.items():
|
||||||
|
assert txn.meta[key] == expected
|
||||||
|
assert txn.postings[0].meta['receipt'] == 'rt:2/13 donation.txt'
|
||||||
|
assert txn.postings[1].meta['receipt'] == 'cash.png rt:2/14'
|
||||||
|
|
Loading…
Reference in a new issue