NPO-Accounting/conservancy_beancount Changeset - c4ce59da753a

Changeset - c4ce59da753a

Parent rev.

Child rev.

[Not reviewed]

0 2 0

Brett Smith - 4 years ago 2020-03-25 03:30:31
brettcsmith@brettcsmith.org

config: Add Config.rt_wrapper method.

This returns a cached version of the new rtutil.RT class,
so that a single process can easily grab references to one
instance with one cache of ticket/attachment data, etc.

2 files changed with 65 insertions and 0 deletions:

conservancy_beancount/config.py

tests/test_config.py

0 comments (0 inline, 0 general)

conservancy_beancount/config.py

➞

Show inline comments

@@ ... / @@ -14,6 +14,7 @@ @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 import functools
 import os
 import urllib.parse as urlparse
@@ ... / @@ -27,6 +28,8 @@ from typing import ( @@
     Type,
+)
 from . import rtutil
 class RTCredentials(NamedTuple):
     server: Optional[str] = None
     user: Optional[str] = None
@@ ... / @@ -99,3 +102,19 @@ class Config: @@
             return retval
         else:
             return None
     @functools.lru_cache(4)
     def _rt_wrapper(self, credentials: RTCredentials, client: Type[rt.Rt]) -> Optional[rtutil.RT]:
         wrapper_client = self.rt_client(credentials, client)
         if wrapper_client is None:
             return None
         else:
             return rtutil.RT(wrapper_client)
     def rt_wrapper(self,
                   credentials: RTCredentials=None,
                   client: Type[rt.Rt]=rt.Rt,
     ) -> Optional[rtutil.RT]:
         if credentials is None:
             credentials = self.rt_credentials()
         return self._rt_wrapper(credentials, client)

tests/test_config.py

➞

Show inline comments

@@ ... / @@ -158,3 +158,49 @@ def test_no_rt_client_without_server(): @@
     rt_credentials = RT_GENERIC_CREDS._replace(server=None, auth='rt')
     config = config_mod.Config()
     assert config.rt_client(rt_credentials, testutil.RTClient) is None
 def test_rt_wrapper():
     config = config_mod.Config()
     rt = config.rt_wrapper(RT_GENERIC_CREDS._replace(auth='rt'), testutil.RTClient)
     assert rt.exists(1)
 def test_rt_wrapper_default_creds():
     config = config_mod.Config()
     rt = config.rt_wrapper(None, testutil.RTClient)
     assert rt.rt.url.startswith(RT_FILE_CREDS[0])
 def test_rt_wrapper_default_creds_from_environ(rt_environ):
     with update_environ(**rt_environ):
         config = config_mod.Config()
         rt = config.rt_wrapper(None, testutil.RTClient)
     assert rt.rt.url.startswith(RT_ENV_CREDS[0])
 def test_rt_wrapper_no_creds():
     with update_environ(HOME=testutil.TESTS_DIR):
         config = config_mod.Config()
         assert config.rt_wrapper(None, testutil.RTClient) is None
 def test_rt_wrapper_bad_creds():
     rt_credentials = RT_GENERIC_CREDS._replace(passwd='badpass', auth='rt')
     config = config_mod.Config()
     assert config.rt_wrapper(rt_credentials, testutil.RTClient) is None
 def test_rt_wrapper_caches():
     rt_credentials = RT_GENERIC_CREDS._replace(auth='rt')
     config = config_mod.Config()
     rt1 = config.rt_wrapper(rt_credentials, testutil.RTClient)
     rt2 = config.rt_wrapper(rt_credentials, testutil.RTClient)
     assert rt1 is rt2
 def test_rt_wrapper_caches_by_creds():
     config = config_mod.Config()
     rt1 = config.rt_wrapper(RT_GENERIC_CREDS._replace(auth='rt'), testutil.RTClient)
     rt2 = config.rt_wrapper(None, testutil.RTClient)
     assert rt1 is not rt2
 def test_rt_wrapper_cache_responds_to_external_credential_changes(rt_environ):
     config = config_mod.Config()
     rt1 = config.rt_wrapper(None, testutil.RTClient)
     with update_environ(**rt_environ):
         rt2 = config.rt_wrapper(None, testutil.RTClient)
     assert rt1 is not rt2

0 comments (0 inline, 0 general)