More tests

jupyterhub/tests/test_app.py

@@ -15,10 +15,11 @@ import pytest
 import traitlets
 from traitlets.config import Config
 
-from .. import orm
+from .. import orm, roles
 from ..app import COOKIE_SECRET_BYTES, JupyterHub
 from .mocking import MockHub
 from .test_api import add_user
+from .utils import api_request, auth_header
 
 
 def test_help_all():
@@ -473,3 +474,107 @@ async def test_user_creation(tmpdir, request):
         "in-group",
         "in-role",
     }
+
+@pytest.fixture(scope='module')
+def db_temp_path(tmp_path_factory):
+    fn = tmp_path_factory.mktemp("db") / "jupyterhub.sqlite"
+    return fn
+
+
+async def test_add_service_at_runtime(request, db_temp_path, service_data):
+    if not os.getenv('JUPYTERHUB_TEST_DB_URL'):
+        p = patch.dict(
+            os.environ,
+            {'JUPYTERHUB_TEST_DB_URL': 'sqlite:///%s' % str(db_temp_path)},
+        )
+        p.start()
+        request.addfinalizer(p.stop)
+
+    kwargs = {"test_clean_db": False}
+    ssl_enabled = getattr(request.module, "ssl_enabled", False)
+    if ssl_enabled:
+        kwargs['internal_certs_location'] = db_temp_path.parents[0]
+    app = MockHub(**kwargs)
+
+    def end():
+        app.log.handlers = []
+        MockHub.clear_instance()
+        try:
+            app.stop()
+        except Exception as e:
+            print("Error stopping Hub: %s" % e, file=sys.stderr)
+
+    request.addfinalizer(end)
+
+    await app.initialize([])
+    await app.start()
+    db = app.db
+
+    user_name = 'admin_services'
+    service_role = {
+        'name': 'admin-services-role',
+        'description': '',
+        'users': [user_name],
+        'scopes': ['admin:services'],
+    }
+    roles.create_role(app.db, service_role)
+    user = add_user(app.db, name=user_name)
+    roles.update_roles(app.db, user, roles=['admin-services-role'])
+
+    service_name = 'service-from-api'
+
+    r = await api_request(
+        app,
+        f'services/{service_name}',
+        headers=auth_header(db, user_name),
+        data=json.dumps(service_data),
+        method='post',
+    )
+    assert r.status_code == 201
+    assert r.json()['name'] == service_name
+    oauth_client = (
+        app.db.query(orm.OAuthClient)
+        .filter_by(identifier=service_data['oauth_client_id'])
+        .first()
+    )
+    assert oauth_client.redirect_uri == service_data['oauth_redirect_uri']
+
+
+async def test_recreate_service_from_database(request, db_temp_path, service_data):
+    if not os.getenv('JUPYTERHUB_TEST_DB_URL'):
+        p = patch.dict(
+            os.environ,
+            {'JUPYTERHUB_TEST_DB_URL': 'sqlite:///%s' % str(db_temp_path)},
+        )
+        p.start()
+        request.addfinalizer(p.stop)
+    kwargs = {"test_clean_db": False}
+
+    ssl_enabled = getattr(request.module, "ssl_enabled", False)
+    if ssl_enabled:
+        kwargs['internal_certs_location'] = db_temp_path.parents[0]
+    app = MockHub(**kwargs)
+
+    def end():
+        app.log.handlers = []
+        MockHub.clear_instance()
+        try:
+            app.stop()
+        except Exception as e:
+            print("Error stopping Hub: %s" % e, file=sys.stderr)
+
+    request.addfinalizer(end)
+
+    await app.initialize([])
+    await app.start()
+
+    assert 'service-from-api' in app._service_map
+    assert (
+        service_data['oauth_client_id'] in app.tornado_settings['oauth_no_confirm_list']
+    )
+    oauth_client = (
+        app.db.query(orm.OAuthClient)
+        .filter_by(identifier=service_data['oauth_client_id'])
+        .first()
+    )
+    assert oauth_client.redirect_uri == service_data['oauth_redirect_uri']