# Standard imports from logging import log from sys import stdout from unittest import skip from unittest.mock import patch from datetime import datetime import subprocess,time,os import multiprocessing # Django imports from django.http import response from django.test import TestCase from django.utils import timezone, tree from django.urls import reverse from django.core import mail from django.conf import settings # Project imports from user_manager.models import PyrosUser,ScientificProgram,Period,SP_Period,SP_Period_User,SP_Period_Guest,UserLevel,Institute,Country from .AgentSP import * # Create your tests here. class ScientificProgramTests(TestCase): fixtures = ['tests/scientific_program_TZ.json'] def setUp(self) -> None: password = "password123" Period.objects.create() self.period2 = Period.objects.create(start_date=Period.objects.all().first().end_date) self.period3 = Period.objects.create(start_date=self.period2.end_date,submission_duration=550) self.usr1 = PyrosUser.objects.get(username="observer") self.usr2 = PyrosUser.objects.get(username="unit_pi") self.usr3 = PyrosUser.objects.get(username="tac") self.usr4 = PyrosUser.objects.get(username="operator") self.usr5 = PyrosUser.objects.get(username="observer2") self.usr6 = PyrosUser.objects.get(username="tac2") self.usr7 = PyrosUser.objects.get(username="observer3") self.usr8 = PyrosUser.objects.get(username="observer4") self.usr1.set_password(password) self.usr1.save() self.usr2.set_password(password) self.usr2.save() self.usr3.set_password(password) self.usr3.save() self.usr4.set_password(password) self.usr4.save() self.usr5.set_password(password) self.usr5.save() self.usr6.set_password(password) self.usr6.save() self.usr7.set_password(password) self.usr7.save() self.usr8.set_password(password) self.usr8.save() def logout(self): self.client.post(reverse("user_logout")) print("Log out") def login_as_user(self,number): if number == 1: self.client.post(reverse("login_validation"),{"email":self.usr1,"password":"password123"}) elif number == 2: self.client.post(reverse("login_validation"),{"email":self.usr2,"password":"password123"}) elif number == 3: self.client.post(reverse("login_validation"),{"email":self.usr3,"password":"password123"}) elif number == 4: self.client.post(reverse("login_validation"),{"email":self.usr4,"password":"password123"}) elif number == 5: self.client.post(reverse("login_validation"),{"email":self.usr5,"password":"password123"}) elif number == 6: self.client.post(reverse("login_validation"),{"email":self.usr6,"password":"password123"}) elif number == 7: self.client.post(reverse("login_validation"),{"email":self.usr7,"password":"password123"}) elif number == 8: self.client.post(reverse("login_validation"),{"email":self.usr8,"password":"password123"}) print(f'Log in as {self.client.session["user"]}') def create_dummy_SCP(self,name,user_number): """ Generate a SCP """ self.login_as_user(user_number) path = reverse('create_scientific_program') post_data = { "name":name, "description_short":"sp short desc", "description_long":"sp long desc", "science_theme":1, } response = self.client.post(path,post_data) self.logout() return ScientificProgram.objects.get(name=name) def create_dummy_SCP_Period(self,scientific_program,user_number): self.login_as_user(user_number) path = reverse('create_scientific_program_period',kwargs={"id_SP":scientific_program.id}) post_data = { "period":Period.objects.next_period().id, "public_visibility":SP_Period.VISIBILITY_YES, "quota_minimal":2, "quota_nominal":5, "over_quota_duration":2, "token":2 } response = self.client.post(path,post_data) sp_period = SP_Period.objects.get(scientific_program=scientific_program,period=Period.objects.next_period()) self.logout() return sp_period def create_dummies_SCP_and_SCP_Period(self,name_of_SCP,user_number): self.create_dummy_SCP(name_of_SCP,user_number) scientific_program = ScientificProgram.objects.get(name=name_of_SCP) sp_period = self.create_dummy_SCP_Period(scientific_program,user_number) return sp_period def test_SCP_Observer_can_submit_new_SP_from_scratch(self): self.login_as_user(1) path = reverse('create_scientific_program') post_data = { "name":"test", "description_short":"sp short desc", "description_long":"sp long desc", "science_theme":1, } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) self.assertEqual(ScientificProgram.objects.all().count(),1) current_sp = ScientificProgram.objects.all().first() self.assertEqual(current_sp.sp_pi,self.usr1) self.logout() self.login_as_user(2) path = reverse('create_scientific_program') post_data = { "name":"test 2", "description_short":"test2 short desc", "description_long":"test2 long desc", "science_theme":2, } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) # user can't access to this page self.assertEqual(response.context,None) self.logout() def test_SCP_Observer_can_submit_new_SP_from_template(self): self.create_dummy_SCP("test",1) current_sp = ScientificProgram.objects.all().first() self.login_as_user(1) path = reverse('create_scientific_program_with_template',kwargs={"id_SP":current_sp.id}) post_data = { "name":"test 2", "description_short":"test2 short desc", "description_long":"test2 long desc", "science_theme":2, } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) self.assertEqual(ScientificProgram.objects.all().count(),2) current_sp = ScientificProgram.objects.all().first() self.assertEqual(current_sp.sp_pi,self.usr1) self.logout() self.login_as_user(2) path = reverse('create_scientific_program_with_template',kwargs={"id_SP":current_sp.id}) post_data = { "name":"test 3", "description_short":"test3 short desc", "description_long":"test3 long desc", "science_theme":2, } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) # user can't access to this page self.assertEqual(response.context,None) self.assertEqual(ScientificProgram.objects.all().count(),2) self.logout() def test_SCP_Observer_can_submit_SP_Period_from_scratch(self): self.create_dummy_SCP("test",1) self.logout() self.login_as_user(1) current_sp = ScientificProgram.objects.all().first() path = reverse('create_scientific_program_period',kwargs={"id_SP":current_sp.id}) post_data = { "period":Period.objects.next_period().id, "public_visibility":SP_Period.VISIBILITY_YES, "quota_minimal":2, "quota_nominal":5, "over_quota_duration":2, "token":2 } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) self.assertEqual(SP_Period.objects.filter(scientific_program=current_sp).count(),1) current_sp_period = SP_Period.objects.filter(scientific_program=current_sp).first() self.assertEqual(current_sp_period.quota_minimal,2) self.assertEqual(current_sp_period.status,SP_Period.STATUSES_DRAFT) self.logout() self.login_as_user(2) path = reverse('create_scientific_program_period',args=[1]) post_data = { "public_visibility":SP_Period.VISIBILITY_YES, "quota_minimal":2, "quota_nominal":5, "quota_allocated":"", "over_quota_duration":2, "over_quota_duration_allocated":"", "token":2, "token_allocated":"", "vote_referee1":"", "reason_referee1":"", "vote_referee2":"", "reason_referee2":"", "priority":"", "is_valid":False } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) # user can't access to this page self.assertEqual(response.context,None) self.logout() def test_SCP_Observer_can_submit_existing_SP_for_new_Period(self): self.create_dummy_SCP("test",1) current_scientific_program = ScientificProgram.objects.all().first() self.create_dummy_SCP_Period(current_scientific_program,1) sp_period = SP_Period.objects.all().first() self.login_as_user(1) path = reverse('repropose_scientific_program',kwargs={"id_SP":current_scientific_program.id,"id_SP_Period":sp_period.id}) post_data = { "period":self.period3.id, "public_visibility":SP_Period.VISIBILITY_YES, "quota_minimal":2, "quota_nominal":5, "over_quota_duration":2, "token":2 } response = self.client.post(path,post_data) self.assertEqual(SP_Period.objects.filter(scientific_program=current_scientific_program).count(),2) current_sp_period = SP_Period.objects.filter(scientific_program=current_scientific_program)[1] SP_Period_User(SP_Period=current_sp_period,user=self.usr5).save() self.assertEqual(current_sp_period.quota_minimal,2) self.assertEqual(current_sp_period.status,SP_Period.STATUSES_DRAFT) # We should have one user which is from the previous period self.assertEqual(SP_Period_User.objects.filter(SP_Period=current_sp_period).count(),1) def test_SCP_Unit_PI_can_add_Period(self): self.login_as_user(2) path = reverse("create_period") today = timezone.now().date().strftime("%d/%m/%Y") post_data = { "start_date_picker":today, "exploitation_duration":182, "submission_duration":182, "evaluation_duration":31, "validation_duration":5, "notification_duration":10, "property_of_data_duration":365, "data_accessibility_duration":365*10 } response = self.client.post(path,post_data) # This creation should fail, we should have 3 periods self.assertEqual(Period.objects.all().count(),3) self.assertEqual(response.status_code,200) self.assertTrue(response.context["error"]) start_date = Period.objects.latest_period().end_date.strftime("%d/%m/%Y") post_data = { "start_date_picker":start_date, "exploitation_duration":182, "submission_duration":182, "evaluation_duration":31, "validation_duration":5, "notification_duration":10, "property_of_data_duration":365, "data_accessibility_duration":365*10 } response = self.client.post(path,post_data) # This creation should success, we should have 4 periods and should be redirected to period_list self.assertEqual(Period.objects.all().count(),4) self.assertEqual(response.status_code,302) self.logout() def test_SCP_view_main_menu(self): self.create_dummy_SCP("test",1) sp_period = self.create_dummy_SCP_Period(ScientificProgram.objects.all().first(),1) self.login_as_user(1) path = reverse("index_scientific_program") response = self.client.get(path) # u1 should be able to add, view scientific programs and view Periods self.assertTrue(response.context["CAN_SUBMIT_SP"]) self.assertTrue(response.context["CAN_VIEW_SP"]) self.assertTrue(response.context["CAN_VIEW_EXPLOITATION_PERIOD"]) self.assertFalse(response.context["CAN_VIEW_ALL_SP"]) self.assertFalse(response.context["CAN_EVALUATE_SP"]) self.assertFalse(response.context["CAN_VALIDATE_SP"]) self.assertFalse(response.context["CAN_CREATE_EXPLOITATION_PERIOD"]) self.assertFalse(response.context["CAN_ASSOCIATE_TAC_TO_SP"]) self.logout() with patch('django.utils.timezone.now', return_value=datetime.strptime(str(Period.objects.next_period().submission_start_date),"%Y-%m-%d")): # u2 should be able to associate TAC to SP # Submission period self.login_as_user(2) response = self.client.get(path) self.assertTrue(response.context["CAN_ASSOCIATE_TAC_TO_SP"]) self.logout() with patch('django.utils.timezone.now', return_value=datetime.strptime(str(Period.objects.next_period().unit_pi_validation_start_date),"%Y-%m-%d")): self.login_as_user(2) # Validation period sp_period.status = SP_Period.STATUSES_EVALUATED sp_period.save() response = self.client.get(path) self.assertEqual(self.client.session["role"],"Unit-PI") # u2 should be able to do everything outside submitting, evaluate SP and see the obsever's view for sp self.assertTrue(response.context["CAN_VIEW_EXPLOITATION_PERIOD"]) self.assertTrue(response.context["CAN_VIEW_ALL_SP"]) self.assertTrue(response.context["CAN_CREATE_EXPLOITATION_PERIOD"]) self.assertTrue(response.context["CAN_VALIDATE_SP"]) self.assertFalse(response.context["CAN_ASSOCIATE_TAC_TO_SP"]) self.assertFalse(response.context["CAN_VIEW_SP"]) self.assertFalse(response.context["CAN_SUBMIT_SP"]) self.assertFalse(response.context["CAN_EVALUATE_SP"]) self.logout() with patch('django.utils.timezone.now', return_value=datetime.strptime(str(Period.objects.next_period().submission_end_date),"%Y-%m-%d")): self.login_as_user(3) sp_period.status = SP_Period.STATUSES_SUBMITTED sp_period.save() response = self.client.get(path) # u3 should be able to only evaluate SP self.assertTrue(response.context["CAN_EVALUATE_SP"]) self.assertFalse(response.context["CAN_VIEW_EXPLOITATION_PERIOD"]) self.assertFalse(response.context["CAN_VIEW_ALL_SP"]) self.assertFalse(response.context["CAN_CREATE_EXPLOITATION_PERIOD"]) self.assertFalse(response.context["CAN_VALIDATE_SP"]) self.assertFalse(response.context["CAN_VIEW_SP"]) self.assertFalse(response.context["CAN_SUBMIT_SP"]) self.assertFalse(response.context["CAN_ASSOCIATE_TAC_TO_SP"]) self.logout() def test_SCP_view_sp_list_and_detail(self): sp1 = self.create_dummies_SCP_and_SCP_Period("test",1) sp2 = self.create_dummies_SCP_and_SCP_Period("test2",1) sp3 = self.create_dummies_SCP_and_SCP_Period("test3",1) sp4 = self.create_dummies_SCP_and_SCP_Period("test4",1) sp5 = self.create_dummies_SCP_and_SCP_Period("test5",1) sp6 = self.create_dummies_SCP_and_SCP_Period("test6",8) sp7 = self.create_dummies_SCP_and_SCP_Period("test7",7) sp2.status = SP_Period.STATUSES_SUBMITTED sp2.referee2 = self.usr3 sp2.referee1 = self.usr6 sp2.save() sp3.status = SP_Period.STATUSES_EVALUATED sp3.referee1 = self.usr6 sp3.referee2 = self.usr3 sp3.save() sp4.status = SP_Period.STATUSES_ACCEPTED sp4.save() SP_Period_User.objects.create(SP_Period=sp4,user=self.usr5) sp5.status = SP_Period.STATUSES_REJECTED sp6.status = SP_Period.STATUSES_ACCEPTED sp6.save() SP_Period_User.objects.create(SP_Period=sp6,user=self.usr1) sp7.status = SP_Period.STATUSES_ACCEPTED sp7.save() SP_Period_User.objects.create(SP_Period=sp7,user=self.usr5) self.login_as_user(1) path = reverse("own_scientific_program_list") response = self.client.get(path) # Can't compare 2 queryset, it returns an error despite they are equal.. so we transform those queryset into list and compare them. self.assertEqual(list(ScientificProgram.objects.filter(sp_pi=self.usr1).order_by("-id")),list(response.context["sp_of_user"])) path = reverse("detail_scientific_program_period",kwargs={"id_sp":sp1.scientific_program.id,"id_period":sp1.period.id}) response = self.client.get(path) self.assertFalse(response.context["CAN_VIEW_TAC_VOTES"]) self.logout() # login as TAC 1 and put today date to evaluation period with patch('django.utils.timezone.now', return_value=datetime.strptime(str(Period.objects.next_period().submission_end_date),"%Y-%m-%d")): self.login_as_user(3) path = reverse("list_submitted_scientific_program") response = self.client.get(path) self.assertEqual(list(SP_Period.objects.filter(referee2=self.usr3,period=Period.objects.next_period(),status=SP_Period.STATUSES_SUBMITTED)),response.context["list_of_evaluated_sp"]) path = reverse("detail_scientific_program_period",kwargs={"id_sp":sp2.scientific_program.id,"id_period":sp2.period.id}) response = self.client.get(path) self.assertEqual(response.status_code,302) # user can't access to this page self.assertEqual(response.context,None) self.logout() # log in as user observer self.login_as_user(5) path = reverse("scientific_program_list") response = self.client.get(path) sp_of_user = list(("test4","test7")) self.assertListEqual(sp_of_user,list(response.context["sp_of_user"].values_list("name",flat=True))) path = reverse("detail_scientific_program_period",kwargs={"id_sp":sp4.scientific_program.id,"id_period":sp4.period.id}) response = self.client.get(path) self.assertFalse(response.context["CAN_VIEW_TAC_VOTES"]) self.assertFalse(response.context["CAN_SUBMIT_PROPOSAL"]) self.logout() # log in as unit pi self.login_as_user(2) path = reverse("scientific_program_list") response = self.client.get(path) self.assertListEqual(list(ScientificProgram.objects.all().order_by("-id")),list(response.context["scientific_programs"])) self.logout() def test_SCP_list_Periods(self): self.login_as_user(1) path = reverse("period_list") response = self.client.get(path) self.assertContains(response,"(Current period)") self.assertEqual(Period.objects.all().count(),len(response.context["future_periods"])) self.logout() def test_SCP_detail_Period(self): sp1 = self.create_dummies_SCP_and_SCP_Period("test",1) sp2 = self.create_dummies_SCP_and_SCP_Period("test2",1) sp3 = self.create_dummies_SCP_and_SCP_Period("test3",1) sp4 = self.create_dummies_SCP_and_SCP_Period("test4",1) sp5 = self.create_dummies_SCP_and_SCP_Period("test5",1) self.login_as_user(1) period = sp1.period path = reverse("detail_period",kwargs={"id":period.id}) response = self.client.get(path) # we should see the same number of SP_Period (5) that are associated to user 1 self.assertEqual(SP_Period.objects.all().count(),len(response.context["sp_periods"])) self.logout() def test_SCP_edit(self): sp1 = self.create_dummies_SCP_and_SCP_Period("test",1) sp2 = self.create_dummies_SCP_and_SCP_Period("test2",1) sp3 = self.create_dummies_SCP_and_SCP_Period("test3",1) sp4 = self.create_dummies_SCP_and_SCP_Period("test4",1) sp5 = self.create_dummies_SCP_and_SCP_Period("test5",5) sp2.status = SP_Period.STATUSES_SUBMITTED sp2.save() sp3.status = SP_Period.STATUSES_ACCEPTED sp3.save() sp4.status = SP_Period.STATUSES_ACCEPTED sp4.save() sp5.status = SP_Period.STATUSES_ACCEPTED sp5.save() post_data = { "name": "test", "description_short":"modified test", "description_long":"modified test", "science_theme" :"2", "public_visibility":"Yes", "quota_minimal":"0", "quota_nominal":"0", "over_quota_duration":"0", "token":"0", "vote_referee1":"", "reason_referee1":"", "users":"", } self.login_as_user(1) path = reverse("edit_scientific_program_period",kwargs={"id_sp":sp1.scientific_program.id,"id_period":sp1.period.id}) response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) updated_sp = SP_Period.objects.get(id=sp1.id) self.assertEqual(updated_sp.scientific_program.description_short,"modified test") path = reverse("edit_scientific_program_period",kwargs={"id_sp":sp2.scientific_program.id,"id_period":sp2.period.id}) post_data = { "name": "test", "description_short":"modified test2", "description_long":"modified test2", "science_theme" :"2", "public_visibility":"Yes", "quota_minimal":"10", "quota_nominal":"0", "over_quota_duration":"0", "token":"0", "vote_referee1":"", "reason_referee1":"", "users":"", } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) updated_sp = SP_Period.objects.get(id=sp2.id) # user can't modify this SP_Period because the status isn't DRAFT self.assertNotEqual(updated_sp.scientific_program.description_short,"modified test2") self.assertNotEqual(updated_sp.quota_minimal,10) path = reverse("edit_scientific_program_period",kwargs={"id_sp":sp5.scientific_program.id,"id_period":sp5.period.id}) response = self.client.post(path,post_data) # u1 can't edit SP5 because he is not the SP_PI self.assertEqual(response.status_code,302) self.assertNotEqual(updated_sp.scientific_program.description_short,"modified test2") self.assertNotEqual(updated_sp.quota_minimal,10) self.assertEqual(response.url,reverse("detail_scientific_program_period",kwargs={"id_sp":sp5.scientific_program.id,"id_period":sp5.period.id})) self.logout() def test_SCP_add_users(self): sp1 = self.create_dummies_SCP_and_SCP_Period("test",1) post_data = { "name": "test", "description_short":"modified test", "description_long":"modified test", "science_theme" :"2", "public_visibility":"Yes", "quota_minimal":"0", "quota_nominal":"0", "over_quota_duration":"0", "token":"0", "vote_referee1":"", "reason_referee1":"", "users":"observer2@test.com", } self.login_as_user(1) path = reverse("edit_scientific_program_period",kwargs={"id_sp":sp1.scientific_program.id,"id_period":sp1.period.id}) response = self.client.post(path,post_data) self.assertEqual(SP_Period_Guest.objects.filter(SP_Period=sp1).first().email,"observer2@test.com") self.assertEqual(len(mail.outbox),1) self.assertIn("observer2@test.com",mail.outbox[0].recipients()) self.logout() url = reverse("sp_register",kwargs={"id_SP":sp1.scientific_program.id,"id_period":sp1.period.id}) self.login_as_user(5) response = self.client.get(url) self.assertEqual(response.status_code,302) self.assertEqual(response.url,reverse("detail_scientific_program_period",kwargs={"id_sp":sp1.scientific_program.id,"id_period":sp1.period.id})) self.assertEqual(SP_Period_Guest.objects.all().count(),0) self.assertEqual(SP_Period_User.objects.filter(SP_Period=sp1).count(),1) self.assertEqual(SP_Period_User.objects.filter(SP_Period=sp1).first().user,self.usr5) self.logout() def test_SCP_delete(self): sp1 = self.create_dummies_SCP_and_SCP_Period("test",1) sp2 = self.create_dummy_SCP("test2",1) self.login_as_user(1) path = reverse("delete_scientific_program",kwargs={"id_sp":sp1.scientific_program.id}) response = self.client.get(path) # Can't delete this SP because it's linked to a period self.assertEqual(response.status_code,302) self.assertEqual(response.url,reverse("detail_scientific_program", args=[sp1.scientific_program.id])) self.assertEqual(ScientificProgram.objects.all().count(),2) path = reverse("delete_scientific_program",kwargs={"id_sp":sp2.id}) response = self.client.get(path) # Can delete this SP self.assertEqual(response.status_code,302) self.assertEqual(response.url,reverse("index_scientific_program")) self.assertEqual(ScientificProgram.objects.all().count(),1) self.logout() def test_SCP_Period_delete(self): sp1 = self.create_dummies_SCP_and_SCP_Period("test",1) sp2 = self.create_dummies_SCP_and_SCP_Period("test2",1) sp2.status = SP_Period.STATUSES_ACCEPTED sp2.save() self.login_as_user(1) path = reverse("delete_scientific_program_period",kwargs={"id_sp":sp1.scientific_program.id,"id_period":sp1.period.id}) response = self.client.get(path) # Can delete this SP because it's in draft status self.assertEqual(response.status_code,302) self.assertEqual(response.url,reverse("detail_scientific_program", args=[sp1.scientific_program.id])) self.assertEqual(SP_Period.objects.filter(scientific_program=sp1.scientific_program).count(),0) path = reverse("delete_scientific_program_period",kwargs={"id_sp":sp2.scientific_program.id,"id_period":sp2.period.id}) response = self.client.get(path) # Can't delete this SP self.assertEqual(response.status_code,302) self.assertEqual(response.url,reverse("detail_scientific_program", args=[sp2.scientific_program.id])) self.assertEqual(SP_Period.objects.filter(scientific_program=sp2.scientific_program).count(),1) self.logout() def run_agent_SP(self): agent = build_agent(AgentSP) print(agent) agent.run() # test not working due to test db and how agent works @skip def test_SCP_lifecycle(self): sp1 = self.create_dummies_SCP_and_SCP_Period("test",1) sp2 = self.create_dummies_SCP_and_SCP_Period("test2",1) sp3 = self.create_dummies_SCP_and_SCP_Period("test3",1) sp3.scientific_program.is_auto_validated = True sp3.scientific_program.save() os.chdir("./scientific_program/") with patch('django.utils.timezone.now', return_value=datetime.strptime(str(Period.objects.next_period().submission_end_date + relativedelta(days=-10)),"%Y-%m-%d")): # TAC Assignation #p = subprocess.Popen("python3 AgentSP.py test",shell=True) p = multiprocessing.Process(target= self.run_agent_SP,args=()) p.daemon = True p.start() time.sleep(10) p.terminate() for sp_period in SP_Period.objects.all().exclude(id=sp3.id): # sp period should have referees self.assertNotEqual(sp_period.referee1,None) self.assertNotEqual(sp_period.referee2,None) with patch('django.utils.timezone.now', return_value=datetime.strptime(str(Period.objects.next_period().submission_end_date),"%Y-%m-%d")): # Evaluation start p = multiprocessing.Process(target= self.run_agent_SP,args=()) p.daemon = True p.start() time.sleep(10) p.terminate() for sp_period in SP_Period.objects.all().exclude(id=sp3.id): # sp period should have referees self.assertEqual(sp_period.status,SP_Period.STATUSES_SUBMITTED) # sp3 should be valid self.assertEqual(SP_Period.objects.get(id=sp3.id).status,SP_Period.STATUSES_ACCEPTED) self.assertEqual(SP_Period.objects.get(id=sp3.id).is_valid,SP_Period.IS_VALID_ACCEPTED) # log in as tac to vote self.login_as_user(3) path = reverse("edit_scientific_program_period",kwargs={"id_sp":sp1.scientific_program.id,"id_period":sp1.period.id}) post_data = { "name": sp1.scientific_program.name, "description_short":"modified test", "description_long":"modified test", "science_theme" :sp1.scientific_program.science_theme, "public_visibility":"Yes", "quota_minimal":sp1.quota_minimal, "quota_nominal":sp1.quota_nominal, "over_quota_duration":sp1.over_quota_duration, "token":sp1.token, "vote_referee1":SP_Period.VOTES_YES, "reason_referee1":"i agree", "users":"", } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) self.assertEqual(SP_Period.objects.get(id=sp1.id).vote_referee1,SP_Period.VOTES_YES) self.assertEqual(SP_Period.objects.get(id=sp1.id).reason_referee1,"i agree") self.logout() # log in as tac 2 to vote self.login_as_user(6) path = reverse("edit_scientific_program_period",kwargs={"id_sp":sp1.scientific_program.id,"id_period":sp1.period.id}) post_data = { "name": sp1.scientific_program.name, "description_short":"modified test", "description_long":"modified test", "science_theme" :sp1.scientific_program.science_theme, "public_visibility":"Yes", "quota_minimal":sp1.quota_minimal, "quota_nominal":sp1.quota_nominal, "over_quota_duration":sp1.over_quota_duration, "token":sp1.token, "vote_referee2":SP_Period.VOTES_YES, "reason_referee2":"i agree", "users":"", } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) self.assertEqual(SP_Period.objects.get(id=sp1.id).vote_referee2,SP_Period.VOTES_YES) self.assertEqual(SP_Period.objects.get(id=sp1.id).reason_referee2,"i agree") self.logout() with patch('django.utils.timezone.now', return_value=datetime.strptime(str(Period.objects.next_period().unit_pi_validation_start_date),"%Y-%m-%d")): # validation start p = multiprocessing.Process(target= self.run_agent_SP,args=()) p.daemon = True p.start() time.sleep(10) p.terminate() for sp_period in SP_Period.objects.all().exclude(id=sp3.id): # sp period should be evaluated self.assertEqual(sp_period.status,SP_Period.STATUSES_EVALUATED) # log in as unit pi self.login_as_user(2) # valid sp1 path = reverse("edit_scientific_program_period",kwargs={"id_sp":sp1.scientific_program.id,"id_period":sp1.period.id}) post_data = { "name": sp1.scientific_program.name, "description_short":"modified test", "description_long":"modified test", "science_theme" :sp1.scientific_program.science_theme, "public_visibility":"Yes", "quota_minimal":sp1.quota_minimal, "quota_nominal":sp1.quota_nominal, "over_quota_duration":sp1.over_quota_duration, "token":sp1.token, "is_valid":SP_Period.IS_VALID_ACCEPTED, "quota_allocated" : 10, "token_allocated": 5, "over_quota_duration_allocated": 5 } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) self.assertEqual(SP_Period.objects.get(id=sp1.id).quota_allocated,10) self.assertEqual(SP_Period.objects.get(id=sp1.id).is_valid,SP_Period.IS_VALID_ACCEPTED) self.assertEqual(SP_Period.objects.get(id=sp1.id).token_allocated,5) self.assertEqual(SP_Period.objects.get(id=sp1.id).over_quota_duration_allocated,5) # reject sp 2 path = reverse("edit_scientific_program_period",kwargs={"id_sp":sp2.scientific_program.id,"id_period":sp2.period.id}) post_data = { "name": sp2.scientific_program.name, "description_short":"modified test", "description_long":"modified test", "science_theme" :sp2.scientific_program.science_theme, "public_visibility":"Yes", "quota_minimal":sp2.quota_minimal, "quota_nominal":sp2.quota_nominal, "over_quota_duration":sp2.over_quota_duration, "token":sp2.token, "is_valid":SP_Period.IS_VALID_REJECTED, "quota_allocated" : 0, "token_allocated": 0, "over_quota_duration_allocated": 0 } response = self.client.post(path,post_data) self.assertEqual(response.status_code,302) self.assertEqual(SP_Period.objects.get(id=sp2.id).is_valid,SP_Period.IS_VALID_REJECTED) self.assertEqual(SP_Period.objects.get(id=sp2.id).quota_allocated,0) self.assertEqual(SP_Period.objects.get(id=sp2.id).token_allocated,0) self.assertEqual(SP_Period.objects.get(id=sp2.id).over_quota_duration_allocated,0) self.logout() with patch('django.utils.timezone.now', return_value=datetime.strptime(str(Period.objects.next_period().notification_start_date),"%Y-%m-%d")): # validation start p = multiprocessing.Process(target= self.run_agent_SP,args=()) p.daemon = True p.start() time.sleep(10) p.terminate() self.assertEqual(SP_Period.objects.get(id=sp1.id).status,SP_Period.STATUSES_ACCEPTED) self.assertEqual(SP_Period.objects.get(id=sp3.id).status,SP_Period.STATUSES_ACCEPTED) self.assertEqual(SP_Period.objects.get(id=sp2.id).status,SP_Period.STATUSES_REJECTED)