From 4f0798fdb0707e726ef5c7857ea032627bf08db2 Mon Sep 17 00:00:00 2001
From: Sergey Yakubov <sergey.yakubov@desy.de>
Date: Wed, 26 Feb 2020 18:12:33 +0100
Subject: [PATCH] add integration test, compare beamtimes between two
 authorizations

---
 producer/api/python/asapo_producer.pyx.in     |  7 +-
 receiver/src/request_handler_authorize.cpp    | 10 ++-
 receiver/src/request_handler_authorize.h      |  2 +-
 .../test_request_handler_authorizer.cpp       | 29 ++++----
 tests/automatic/producer/CMakeLists.txt       |  2 +
 tests/automatic/producer/aai/CMakeLists.txt   | 30 ++++++++
 .../aai/beamtime-metadata-11111111.json       | 36 +++++++++
 tests/automatic/producer/aai/check_linux.sh   | 52 +++++++++++++
 .../automatic/producer/aai/check_windows.bat  | 61 +++++++++++++++
 tests/automatic/producer/aai/producer_aai.py  | 74 +++++++++++++++++++
 .../producer/aai/settings.json.tpl.in         |  8 ++
 .../settings/receiver.json.tpl.lin.in         |  2 +-
 .../settings/receiver.json.tpl.win.in         |  2 +-
 13 files changed, 289 insertions(+), 26 deletions(-)
 create mode 100644 tests/automatic/producer/aai/CMakeLists.txt
 create mode 100644 tests/automatic/producer/aai/beamtime-metadata-11111111.json
 create mode 100644 tests/automatic/producer/aai/check_linux.sh
 create mode 100644 tests/automatic/producer/aai/check_windows.bat
 create mode 100644 tests/automatic/producer/aai/producer_aai.py
 create mode 100644 tests/automatic/producer/aai/settings.json.tpl.in

diff --git a/producer/api/python/asapo_producer.pyx.in b/producer/api/python/asapo_producer.pyx.in
index e606d7d89..c382900c7 100644
--- a/producer/api/python/asapo_producer.pyx.in
+++ b/producer/api/python/asapo_producer.pyx.in
@@ -278,11 +278,12 @@ cdef class PyProducer:
             if self.c_producer.get() is not NULL:
                 self.c_producer.get().StopThreads__()
     @staticmethod
-    def __create_producer(endpoint,beamtime_id,stream,token,nthreads,timeout_sec):
+    def __create_producer(endpoint,beamtime_id,beamline,stream,token,nthreads,timeout_sec):
         pyProd = PyProducer()
         cdef Error err
         cdef SourceCredentials source
         source.beamtime_id = beamtime_id
+        source.beamline = beamline
         source.user_token = token
         source.stream = stream
         pyProd.c_producer = Producer.Create(endpoint,nthreads,RequestHandlerType_Tcp,source,timeout_sec,&err)
@@ -290,7 +291,7 @@ cdef class PyProducer:
             throw_exception(err)
         return pyProd
 
-def create_producer(endpoint,beamtime_id,stream,token,nthreads,timeout_sec):
+def create_producer(endpoint,beamtime_id='auto',beamline='auto',stream='detector',token='',nthreads=1,timeout_sec=3600):
     """
          :param endpoint: server endpoint (url:port)
          :type endpoint: string
@@ -308,7 +309,7 @@ def create_producer(endpoint,beamtime_id,stream,token,nthreads,timeout_sec):
             AsapoWrongInputError: wrong input (number of threads, ,,,)
             AsapoProducerError: actually should not happen
     """
-    return PyProducer.__create_producer(_bytes(endpoint),_bytes(beamtime_id),_bytes(stream),_bytes(token),nthreads,timeout_sec)
+    return PyProducer.__create_producer(_bytes(endpoint),_bytes(beamtime_id),_bytes(beamline),_bytes(stream),_bytes(token),nthreads,timeout_sec)
 
 
 __version__ = "@ASAPO_VERSION_PYTHON@"
diff --git a/receiver/src/request_handler_authorize.cpp b/receiver/src/request_handler_authorize.cpp
index c1d975c62..9c35a498e 100644
--- a/receiver/src/request_handler_authorize.cpp
+++ b/receiver/src/request_handler_authorize.cpp
@@ -72,8 +72,9 @@ Error RequestHandlerAuthorize::ProcessAuthorizationRequest(Request* request) con
     return Authorize(request, request->GetMessage());
 }
 
-Error RequestHandlerAuthorize::ProcessReAuthorizeError(const Request* request,Error err) const {
-    if (err == asapo::ReceiverErrorTemplates::kAuthorizationFailure && request->GetBeamtimeId() == "auto") {
+Error RequestHandlerAuthorize::ProcessReAuthorization(const Request* request, Error err) const {
+    if (err == asapo::ReceiverErrorTemplates::kAuthorizationFailure || (
+        err==nullptr && request->GetBeamtimeId()!=beamtime_id_)) {
         return asapo::ReceiverErrorTemplates::kReAuthorizationFailure.Generate();
     }
     return err;
@@ -88,8 +89,9 @@ Error RequestHandlerAuthorize::ProcessOtherRequest(Request* request) const {
                           (system_clock::now() - last_updated_).count();
     if (elapsed_ms >= GetReceiverConfig()->authorization_interval_ms) {
         auto err = Authorize(request, cached_source_credentials_.c_str());
-        if (err) {
-            return ProcessReAuthorizeError(request, std::move(err));
+        auto reauth_err = ProcessReAuthorization(request, std::move(err));
+        if (reauth_err) {
+            return reauth_err;
         }
     }
     request->SetBeamtimeId(beamtime_id_);
diff --git a/receiver/src/request_handler_authorize.h b/receiver/src/request_handler_authorize.h
index 1a5edcf52..642a1919c 100644
--- a/receiver/src/request_handler_authorize.h
+++ b/receiver/src/request_handler_authorize.h
@@ -31,7 +31,7 @@ class RequestHandlerAuthorize final: public ReceiverRequestHandler {
     Error ProcessOtherRequest(Request* request) const;
     Error Authorize(Request* request, const char* source_credentials) const;
     Error ErrorFromAuthorizationServerResponse(const Error& err, HttpCode code) const;
-    Error ProcessReAuthorizeError(const Request* request,Error err) const;
+    Error ProcessReAuthorization(const Request* request, Error err) const;
     std::string GetRequestString(const Request* request, const char* source_credentials) const;
 };
 
diff --git a/receiver/unittests/test_request_handler_authorizer.cpp b/receiver/unittests/test_request_handler_authorizer.cpp
index a08a70921..b2a444115 100644
--- a/receiver/unittests/test_request_handler_authorizer.cpp
+++ b/receiver/unittests/test_request_handler_authorizer.cpp
@@ -144,17 +144,12 @@ class AuthorizerHandlerTests : public Test {
         MockAuthRequest(error, code);
         return handler.ProcessRequest(mock_request.get());
     }
-    Error MockRequestAuthorization(bool error, HttpCode code = HttpCode::OK) {
+    Error MockRequestAuthorization(bool error, HttpCode code = HttpCode::OK,std::string return_beamtime_id="beamtime_id") {
         EXPECT_CALL(*mock_request, GetOpCode())
         .WillOnce(Return(asapo::kOpcodeTransferData))
         ;
 
-        if (!error && code == HttpCode::Unauthorized) {
-        EXPECT_CALL(*mock_request, GetBeamtimeId())
-            .WillOnce(ReturnRef(expected_beamtime_id))
-            ;
-        }
-        if (!error && code == HttpCode::OK) {
+        if (!error && code == HttpCode::OK && return_beamtime_id==expected_beamtime_id) {
             EXPECT_CALL(*mock_request, SetBeamtimeId(expected_beamtime_id));
             EXPECT_CALL(*mock_request, SetStream(expected_stream));
             EXPECT_CALL(*mock_request, SetOfflinePath(expected_core_path));
@@ -222,6 +217,7 @@ TEST_F(AuthorizerHandlerTests, ErrorOnSecondAuthorize) {
 
 TEST_F(AuthorizerHandlerTests, ErrorOnDataTransferRequestAuthorize) {
     MockFirstAuthorization(false);
+
     auto err = MockRequestAuthorization(true);
 
     ASSERT_THAT(err, Eq(asapo::ReceiverErrorTemplates::kInternalServerError));
@@ -233,26 +229,27 @@ TEST_F(AuthorizerHandlerTests, DataTransferRequestAuthorizeReturns401) {
 
     auto err = MockRequestAuthorization(false, HttpCode::Unauthorized);
 
-    ASSERT_THAT(err, Eq(asapo::ReceiverErrorTemplates::kAuthorizationFailure));
+    ASSERT_THAT(err, Eq(asapo::ReceiverErrorTemplates::kReAuthorizationFailure));
 }
 
-TEST_F(AuthorizerHandlerTests, DataTransferRequestAuthorizeReturnsOK) {
+TEST_F(AuthorizerHandlerTests, DataTransferRequestAuthorizeReturnsSameBeamtimeId) {
     MockFirstAuthorization(false);
+    EXPECT_CALL(*mock_request, GetBeamtimeId())
+        .WillOnce(ReturnRef(expected_beamtime_id))
+        ;
 
     auto err = MockRequestAuthorization(false);
 
     ASSERT_THAT(err, Eq(nullptr));
 }
 
-TEST_F(AuthorizerHandlerTests, RequestAuthorizeReturnsReqauthorize) {
-    expected_beamtime_id="auto";
-    expected_source_credentials = "auto%stream%token";
-    expect_request_string = std::string("{\"SourceCredentials\":\"") + expected_source_credentials +
-        "\",\"OriginHost\":\"" +expected_producer_uri + "\"}";
-
+TEST_F(AuthorizerHandlerTests, RequestAuthorizeReturnsDifferentBeamtimeId) {
     MockFirstAuthorization(false);
+    EXPECT_CALL(*mock_request, GetBeamtimeId())
+        .WillOnce(ReturnRef("beamtime_id2"))
+        ;
 
-    auto err = MockRequestAuthorization(false,HttpCode::Unauthorized);
+    auto err = MockRequestAuthorization(false,HttpCode::OK,"beamtime_id2");
 
     ASSERT_THAT(err, Eq(asapo::ReceiverErrorTemplates::kReAuthorizationFailure));
 }
diff --git a/tests/automatic/producer/CMakeLists.txt b/tests/automatic/producer/CMakeLists.txt
index 5974fe371..e96695ebb 100644
--- a/tests/automatic/producer/CMakeLists.txt
+++ b/tests/automatic/producer/CMakeLists.txt
@@ -3,4 +3,6 @@ add_subdirectory(beamtime_metadata)
 
 if (UNIX OR CMAKE_BUILD_TYPE STREQUAL "Release")
     add_subdirectory(python_api)
+    add_subdirectory(aai)
+
 endif()
diff --git a/tests/automatic/producer/aai/CMakeLists.txt b/tests/automatic/producer/aai/CMakeLists.txt
new file mode 100644
index 000000000..2330026be
--- /dev/null
+++ b/tests/automatic/producer/aai/CMakeLists.txt
@@ -0,0 +1,30 @@
+set(TARGET_NAME producer_python_aai)
+
+################################
+# Testing
+################################
+prepare_asapo()
+
+prepare_asapo()
+
+if (UNIX)
+    get_target_property(PYTHON_LIBS python-lib-producer BINARY_DIR)
+else()
+    get_target_property(PYTHON_LIBS asapo_producer BINARY_DIR)
+endif()
+
+file(TO_NATIVE_PATH ${CMAKE_CURRENT_BINARY_DIR}/beamline CURRENT_BEAMLINES_FOLDER )
+
+if (WIN32)
+    string(REPLACE "\\" "\\\\" CURRENT_BEAMLINES_FOLDER "${CURRENT_BEAMLINES_FOLDER}")
+endif()
+
+configure_file(${CMAKE_CURRENT_SOURCE_DIR}/settings.json.tpl.in authorizer.json.tpl @ONLY)
+configure_file(beamtime-metadata-11111111.json beamline/p07/current/beamtime-metadata-11111111.json COPYONLY)
+
+file(TO_NATIVE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/producer_aai.py TEST_SCRIPT )
+
+add_script_test("${TARGET_NAME}-python" "${Python_EXECUTABLE} ${PYTHON_LIBS} ${TEST_SCRIPT} " nomem)
+
+
+
diff --git a/tests/automatic/producer/aai/beamtime-metadata-11111111.json b/tests/automatic/producer/aai/beamtime-metadata-11111111.json
new file mode 100644
index 000000000..65820fcb0
--- /dev/null
+++ b/tests/automatic/producer/aai/beamtime-metadata-11111111.json
@@ -0,0 +1,36 @@
+{
+  "applicant": {
+    "email": "test",
+    "institute": "test",
+    "lastname": "test",
+    "userId": "1234",
+    "username": "test"
+  },
+  "beamline": "p07",
+  "beamline_alias": "P07",
+  "beamtimeId": "11111111",
+  "contact": "None",
+  "core-path": "/tmp/asapo/receiver/files/test_facility/gpfs/p07/2019/data/11111111",
+  "event-end": "2020-03-03 09:00:00",
+  "event-start": "2020-03-02 09:00:00",
+  "facility": "PETRA III",
+  "generated": "2020-02-22 22:37:16",
+  "pi": {
+    "email": "test",
+    "institute": "test",
+    "lastname": "test",
+    "userId": "14",
+    "username": "test"
+  },
+  "proposalId": "12345678",
+  "proposalType": "H",
+  "title": "In-House Research (P07)",
+  "unixId": "None",
+  "users": {
+    "door-db": [
+      "test"
+    ],
+    "special": [],
+    "unknown": []
+  }
+}
diff --git a/tests/automatic/producer/aai/check_linux.sh b/tests/automatic/producer/aai/check_linux.sh
new file mode 100644
index 000000000..826762621
--- /dev/null
+++ b/tests/automatic/producer/aai/check_linux.sh
@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+
+#set -e
+
+trap Cleanup EXIT
+
+beamtime_id=11111111
+beamtime_id2=22222222
+stream=python
+beamline=p07
+receiver_root_folder=/tmp/asapo/receiver/files
+facility=test_facility
+year=2019
+receiver_folder=${receiver_root_folder}/${facility}/gpfs/${beamline}/${year}/data/${beamtime_id}
+receiver_folder2=${receiver_root_folder}/${facility}/gpfs/${beamline}/${year}/data/${beamtime_id2}
+token=-pZmisCNjAbjT2gFBKs3OB2kNOU79SNsfHud0bV8gS4= # for bl_p07
+
+
+Cleanup() {
+	echo cleanup
+	rm -rf ${receiver_root_folder}
+    nomad stop receiver >/dev/null
+    nomad stop discovery >/dev/null
+    nomad stop authorizer >/dev/null
+    nomad stop nginx >/dev/null
+    nomad run nginx_kill.nmd  && nomad stop -yes -purge nginx_kill > /dev/null
+    echo "db.dropDatabase()" | mongo ${beamtime_id}_${stream} >/dev/null
+    echo "db.dropDatabase()" | mongo ${beamtime_id2}_${stream} >/dev/null
+}
+
+export PYTHONPATH=$2:${PYTHONPATH}
+
+echo "db.${beamtime_id}_${stream}.insert({dummy:1})" | mongo ${beamtime_id}_${stream}  >/dev/null
+
+nomad run authorizer.nmd >/dev/null
+nomad run nginx.nmd >/dev/null
+nomad run receiver.nmd >/dev/null
+nomad run discovery.nmd >/dev/null
+
+mkdir -p ${receiver_folder} ${receiver_folder2}
+
+sleep 1
+
+echo test > file1
+
+
+$1 $3 $beamline $token $stream "127.0.0.1:8400" > out || cat out
+cat out
+cat out | grep "successfuly sent" | wc -l | grep 4
+cat out | grep "reauthorization" | wc -l | grep 1
+cat out | grep "duplicated" | wc -l | grep 2
+
diff --git a/tests/automatic/producer/aai/check_windows.bat b/tests/automatic/producer/aai/check_windows.bat
new file mode 100644
index 000000000..2acbb41ac
--- /dev/null
+++ b/tests/automatic/producer/aai/check_windows.bat
@@ -0,0 +1,61 @@
+SET mongo_exe="c:\Program Files\MongoDB\Server\4.2\bin\mongo.exe"
+SET beamtime_id=11111111
+SET beamtime_id2=22222222
+SET beamline=p07
+SET stream=python
+SET receiver_root_folder=c:\tmp\asapo\receiver\files
+SET receiver_folder="%receiver_root_folder%\test_facility\gpfs\%beamline%\2019\data\%beamtime_id%"
+SET receiver_folder2="%receiver_root_folder%\test_facility\gpfs\%beamline%\2019\data\%beamtime_id2%"
+SET dbname=%beamtime_id%_%stream%
+SET dbname2=%beamtime_id2%_%stream%
+SET token=-pZmisCNjAbjT2gFBKs3OB2kNOU79SNsfHud0bV8gS4=
+
+echo db.%dbname%.insert({dummy:1})" | %mongo_exe% %dbname%
+
+c:\opt\consul\nomad run receiver.nmd
+c:\opt\consul\nomad run authorizer.nmd
+c:\opt\consul\nomad run discovery.nmd
+c:\opt\consul\nomad run nginx.nmd
+
+ping 1.0.0.0 -n 5 -w 100 > nul
+
+mkdir %receiver_folder%
+mkdir %receiver_folder2%
+
+echo test > file1
+
+ping 1.0.0.0 -n 1 -w 100 > nul
+
+set PYTHONPATH=%2
+
+"%1" "%3" %beamline% %token%  %stream% "127.0.0.1:8400" > out
+
+type out
+set NUM=0
+for /F %%N in ('find /C "successfuly sent" ^< "out"') do set NUM=%%N
+echo %NUM% | findstr 4 || goto error
+
+for /F %%N in ('find /C "reauthorization" ^< "out"') do set NUM=%%N
+echo %NUM% | findstr 1 || goto error
+
+for /F %%N in ('find /C "} server warning: duplicated request" ^< "out"') do set NUM=%%N
+echo %NUM% | findstr 1 || goto error
+
+goto :clean
+
+:error
+call :clean
+exit /b 1
+
+:clean
+c:\opt\consul\nomad stop receiver
+c:\opt\consul\nomad stop discovery
+c:\opt\consul\nomad stop nginx
+c:\opt\consul\nomad run nginx_kill.nmd  && c:\opt\consul\nomad stop -yes -purge nginx_kill
+c:\opt\consul\nomad stop authorizer
+rmdir /S /Q %receiver_root_folder%
+rmdir /S /Q %receiver_root_folder2%
+echo db.dropDatabase() | %mongo_exe% %dbname%
+echo db.dropDatabase() | %mongo_exe% %dbname2%
+
+
diff --git a/tests/automatic/producer/aai/producer_aai.py b/tests/automatic/producer/aai/producer_aai.py
new file mode 100644
index 000000000..57b3696cd
--- /dev/null
+++ b/tests/automatic/producer/aai/producer_aai.py
@@ -0,0 +1,74 @@
+from __future__ import print_function
+
+import asapo_producer
+import sys
+import time
+import numpy as np
+import threading
+lock = threading.Lock()
+import json
+
+beamline = sys.argv[1]
+token = sys.argv[2]
+stream = sys.argv[3]
+endpoint = sys.argv[4]
+
+nthreads = 1
+
+def callback(header,err):
+    lock.acquire() # to print
+    if isinstance(err,asapo_producer.AsapoServerWarning):
+        print("successfuly sent, but with warning from server: ",header,err)
+    elif err is not None:
+        print("could not sent: ",header,err)
+    else:
+        print ("successfuly sent: ",header)
+    lock.release()
+
+
+producer  = asapo_producer.create_producer(endpoint,beamline=beamline, stream=stream, token=token, nthreads=nthreads, timeout_sec=60)
+
+producer.set_log_level("debug")
+
+#send single file
+producer.send_file(1, local_path = "./file1", exposed_path = stream+"/"+"file1", user_meta = '{"test_key":"test_val"}', callback = callback)
+
+producer.wait_requests_finished(10000)
+
+
+#send single file to other beamtime - should be warning on duplicated request
+producer.send_file(1, local_path = "./file1", exposed_path = stream+"/"+"file1", user_meta = '{"test_key":"test_val"}', callback = callback)
+producer.wait_requests_finished(10000)
+
+fname = 'beamline/p07/current/beamtime-metadata-11111111.json'
+with open(fname) as json_file:
+    data = json.load(json_file)
+data['beamtimeId']='22222222'
+data['core-path']=data['core-path'].replace('11111111','22222222')
+
+with open(fname, 'w') as outfile:
+    json.dump(data, outfile)
+
+time.sleep(2)
+
+#send single file to other beamtime - now ok since receiver authorization timed out
+producer.send_file(1, local_path = "./file1", exposed_path = stream+"/"+"file1", user_meta = '{"test_key":"test_val"}', callback = callback)
+
+producer.wait_requests_finished(10000)
+
+data['beamtimeId']='11111111'
+data['core-path']=data['core-path'].replace('22222222','11111111')
+
+with open(fname, 'w') as outfile:
+    json.dump(data, outfile)
+
+
+n = producer.get_requests_queue_size()
+if n!=0:
+    print("number of remaining requests should be zero, got ",n)
+    sys.exit(1)
+
+
+
+
+
diff --git a/tests/automatic/producer/aai/settings.json.tpl.in b/tests/automatic/producer/aai/settings.json.tpl.in
new file mode 100644
index 000000000..319ef7063
--- /dev/null
+++ b/tests/automatic/producer/aai/settings.json.tpl.in
@@ -0,0 +1,8 @@
+{
+  "Port": {{ env "NOMAD_PORT_authorizer" }},
+  "LogLevel":"debug",
+  "CurrentBeamlinesFolder":"@CURRENT_BEAMLINES_FOLDER@",
+  "SecretFile":"auth_secret.key"
+}
+
+
diff --git a/tests/automatic/settings/receiver.json.tpl.lin.in b/tests/automatic/settings/receiver.json.tpl.lin.in
index e21a4b403..b91057933 100644
--- a/tests/automatic/settings/receiver.json.tpl.lin.in
+++ b/tests/automatic/settings/receiver.json.tpl.lin.in
@@ -14,7 +14,7 @@
     "ReservedShare": 10
   },
   "AuthorizationServer": "localhost:8400/authorizer",
-  "AuthorizationInterval": 10000,
+  "AuthorizationInterval": 1000,
   "ListenPort": {{ env "NOMAD_PORT_recv" }},
   "Tag": "{{ env "NOMAD_ADDR_recv" }}",
   "WriteToDisk": @RECEIVER_WRITE_TO_DISK@,
diff --git a/tests/automatic/settings/receiver.json.tpl.win.in b/tests/automatic/settings/receiver.json.tpl.win.in
index 6cc7aa95d..f46bd5bdf 100644
--- a/tests/automatic/settings/receiver.json.tpl.win.in
+++ b/tests/automatic/settings/receiver.json.tpl.win.in
@@ -5,7 +5,7 @@
   "DatabaseServer":"auto",
   "DiscoveryServer": "localhost:8400/discovery",
   "AuthorizationServer": "localhost:8400/authorizer",
-  "AuthorizationInterval": 10000,
+  "AuthorizationInterval": 1000,
   "ListenPort": {{ env "NOMAD_PORT_recv" }},
   "DataServer": {
     "NThreads": 2,
-- 
GitLab