from logging import log from sys import stdout from unittest import skip from django.http import response from django.test import TestCase from django.utils import timezone, tree from common.models import PyrosUser,ScientificProgram,Period,SP_Period,SP_Period_User,SP_Period_Guest,UserLevel,Institute,Country from django.urls import reverse from django.core import mail from django.conf import settings from datetime import date from datetime import datetime from unittest.mock import patch import subprocess,time,os import multiprocessing 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.get(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)