Merge pull request #2434 from alphagov/get-org-by-domain

Add a method to guess organisation from a domain

app/organisation/rest.py

@@ -1,10 +1,11 @@
-from flask import Blueprint, jsonify, request, current_app
+from flask import abort, Blueprint, jsonify, request, current_app
 from sqlalchemy.exc import IntegrityError
 
 from app.dao.organisation_dao import (
     dao_create_organisation,
     dao_get_organisations,
     dao_get_organisation_by_id,
+    dao_get_organisation_by_email_address,
     dao_get_organisation_services,
     dao_update_organisation,
     dao_add_service_to_organisation,
@@ -53,6 +54,24 @@ def get_organisation_by_id(organisation_id):
     return jsonify(organisation.serialize())
 
 
+@organisation_blueprint.route('/by-domain', methods=['GET'])
+def get_organisation_by_domain():
+
+    domain = request.args.get('domain')
+
+    if not domain or '@' in domain:
+        abort(400)
+
+    organisation = dao_get_organisation_by_email_address(
+        'example@{}'.format(request.args.get('domain'))
+    )
+
+    if not organisation:
+        abort(404)
+
+    return jsonify(organisation.serialize())
+
+
 @organisation_blueprint.route('', methods=['POST'])
 def create_organisation():
     data = request.get_json()

tests/app/organisation/test_rest.py

@@ -85,6 +85,41 @@ def test_get_organisation_by_id_returns_domains(admin_request, notify_db_session
     }
 
 
+@pytest.mark.parametrize('domain, expected_status', (
+    ('foo.gov.uk', 200),
+    ('bar.gov.uk', 200),
+    ('oof.gov.uk', 404),
+    pytest.param(
+        'rab.gov.uk', 200,
+        marks=pytest.mark.xfail(raises=AssertionError),
+    ),
+    (None, 400),
+    ('personally.identifying.information@example.com', 400),
+))
+def test_get_organisation_by_domain(
+    admin_request,
+    notify_db_session,
+    domain,
+    expected_status
+):
+    org = create_organisation()
+    other_org = create_organisation('Other organisation')
+    create_domain('foo.gov.uk', org.id)
+    create_domain('bar.gov.uk', org.id)
+    create_domain('rab.gov.uk', other_org.id)
+
+    response = admin_request.get(
+        'organisation.get_organisation_by_domain',
+        _expected_status=expected_status,
+        domain=domain,
+    )
+
+    if expected_status == 200:
+        assert response['id'] == str(org.id)
+    else:
+        assert response['result'] == 'error'
+
+
 def test_post_create_organisation(admin_request, notify_db_session):
     data = {
         'name': 'test organisation',