import os import unittest import mockredis from nose.tools import eq_ from numpy.testing import assert_array_equal from helix.constants.anchor_type import AnchorType from helix.constants.inverter_type import InverterType from helix.constants.module_type import ModuleType from helix.constants.system_type import SystemType from helix.models.sql.inverters import Inverter from helix.models.sql.power_monitors import PowerMonitor from helix.models.sql.power_stations import PowerStation from helix.models.sql.sites import Site from helix.models.sql.standalone_inverters import StandaloneInverter from helix.models.sql.users import User from test.test_helpers import test_db_session from helix.session_manager import SessionManager class SessionManagerTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.test_db_session = test_db_session() @classmethod def tearDownClass(cls): cls.test_db_session.close_all() def setUp(self): self.redis_store = mockredis.mock_redis_client() self.session = {'id': 'abcdef'} self.test_db_session.rollback() self.test_db_session.query(Inverter).delete() self.test_db_session.query(PowerMonitor).delete() self.test_db_session.query(StandaloneInverter).delete() self.test_db_session.query(PowerStation).delete() self.test_db_session.query(Site).delete() self.test_db_session.query(User).delete() self.test_db_session.commit() self.user = User(username='abcdef', password_hash='') self.test_db_session.add(self.user) self.test_db_session.commit() self.site = Site( user_id=self.user.id, project_name='Foo', building_height=100, building_width=100, building_length=100, parapet_height=100, wind_speed=100, exposure_category='B', exposure_transition_distance=5, ballast_block_weight=12, max_psf=12, system_type=SystemType.dualTilt.value, module_type=ModuleType.Cell128.value, anchor_type=AnchorType.OMG_PowerGrip_Plus.value, spectral_response=2, seismic_importance_factor=1, ) self.test_db_session.add(self.site) self.test_db_session.commit() def test_creates_new_uuid_if_it_doesnt_exist(self): fresh_session = {} SessionManager(fresh_session, None, self.test_db_session) assert 'id' in fresh_session def test_doesnt_change_id_if_it_exists(self): SessionManager(self.session, None, self.test_db_session) eq_(self.session['id'], 'abcdef') def test_does_not_create_multiple_users_with_same_username(self): eq_(self.test_db_session.query(User).count(), 1) SessionManager(self.session, None, self.test_db_session) eq_(self.test_db_session.query(User).count(), 1) def test_creates_new_user_if_does_not_exist(self): self.test_db_session.query(Site).delete() self.test_db_session.query(User).delete() eq_(self.test_db_session.query(User).count(), 0) SessionManager(self.session, None, self.test_db_session) eq_(self.test_db_session.query(User).count(), 1) def test_generates_context_steps_with_valid_site_data(self): session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context(), { 'id': 'abcdef', 'google_analytics_token': '', 'csv_available': False, 'site_data_available': True, 'inverter_brands': [], 'power_stations': [], 'standalone_inverters': [], 'power_monitors': [], 'system_type': SystemType.dualTilt, 'steps': [ (1, 'Site Characterization', '/site_characterization/', True), (2, 'Site Summary', '/summary/', True), (3, 'Array Summary', '/array_summary/', False), (4, 'Power Station Configuration', '/power_station_configuration/', False), (5, 'Download', '/download/', None) ], }) def test_generates_context_without_site_data_outputs_no_site_characterization_steps(self): self.test_db_session.query(Site).delete() session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context(), { 'id': 'abcdef', 'google_analytics_token': '', 'csv_available': False, 'site_data_available': False, 'inverter_brands': [], 'power_stations': [], 'standalone_inverters': [], 'power_monitors': [], 'system_type': SystemType.dualTilt, 'steps': [ (1, 'Site Characterization', '/site_characterization/', False), (2, 'Site Summary', '/summary/', False), (3, 'Array Summary', '/array_summary/', False), (4, 'Power Station Configuration', '/power_station_configuration/', False), (5, 'Download', '/download/', None) ], }) def test_generates_context_with_cad_file_outputs_csv_available(self): self.site.cad_file = 'a,b,c,d' self.test_db_session.commit() session_manager = SessionManager(self.session, self.redis_store, self.test_db_session) eq_(session_manager.context(), { 'id': 'abcdef', 'google_analytics_token': '', 'csv_available': True, 'site_data_available': True, 'inverter_brands': [], 'power_stations': [], 'standalone_inverters': [], 'power_monitors': [], 'system_type': SystemType.dualTilt, 'steps': [ (1, 'Site Characterization', '/site_characterization/', True), (2, 'Site Summary', '/summary/', True), (3, 'Array Summary', '/array_summary/', True), (4, 'Power Station Configuration', '/power_station_configuration/', False), (5, 'Download', '/download/', None) ], }) def test_generates_context_with_analytics_envvar_outputs_analytics_token(self): self.test_db_session.commit() session_manager = SessionManager(self.session, self.redis_store, self.test_db_session) os.environ['GOOGLE_ANALYTICS'] = 'Hello' eq_(session_manager.context(), { 'id': 'abcdef', 'google_analytics_token': 'Hello', 'csv_available': False, 'site_data_available': True, 'inverter_brands': [], 'power_stations': [], 'standalone_inverters': [], 'power_monitors': [], 'system_type': SystemType.dualTilt, 'steps': [ (1, 'Site Characterization', '/site_characterization/', True), (2, 'Site Summary', '/summary/', True), (3, 'Array Summary', '/array_summary/', False), (4, 'Power Station Configuration', '/power_station_configuration/', False), (5, 'Download', '/download/', None) ], }) os.environ['GOOGLE_ANALYTICS'] = '' def test_generates_context_with_site_data_for_single_tilt(self): self.site.system_type = SystemType.singleTilt.value self.test_db_session.commit() session_manager = SessionManager(self.session, self.redis_store, self.test_db_session) eq_(session_manager.context(), { 'id': 'abcdef', 'google_analytics_token': '', 'csv_available': False, 'site_data_available': True, 'inverter_brands': [], 'power_stations': [], 'standalone_inverters': [], 'power_monitors': [], 'system_type': SystemType.singleTilt, 'steps': [ (1, 'Site Characterization', '/site_characterization/', True), (2, 'Site Summary', '/summary/', True), (3, 'Array Summary', '/array_summary/', False), (4, 'Power Station Configuration', '/power_station_configuration/', False), (5, 'Download', '/download/', None) ], }) def test_generates_context_with_power_stations_outputs_power_station_config_step(self): power_station = PowerStation( site_id=self.site.id, quantity=1, ac_run_length=10, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() session_manager = SessionManager(self.session, None, self.test_db_session) context = session_manager.context() eq_(context['steps'][3], (4, 'Power Station Configuration', '/power_station_configuration/', True)) def test_generates_context_with_standalone_inverters_outputs_power_station_config_step(self): power_station = PowerStation( site_id=self.site.id, quantity=1, ac_run_length=10, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() standalone_inverter = StandaloneInverter( site_id=self.site.id, attachment_point_id=power_station.id, ac_run_length=10, ) self.test_db_session.add(standalone_inverter) self.test_db_session.commit() inverter = Inverter( model=str(InverterType.SMA.MODEL_12KW.value), strings_per_inverter=8, sunshade=False, dc_switch=False, power_station_id=None, standalone_inverter_id=standalone_inverter.id ) self.test_db_session.add(inverter) self.test_db_session.commit() session_manager = SessionManager(self.session, None, self.test_db_session) context = session_manager.context() eq_(context['steps'][3], (4, 'Power Station Configuration', '/power_station_configuration/', True)) def test_save_form_submission_creates_new_project_site_if_none_exists_for_user(self): self.test_db_session.query(Site).delete() self.test_db_session.query(User).delete() eq_(self.test_db_session.query(Site).count(), 0) session_manager = SessionManager(self.session, None, self.test_db_session) form_data = { 'project_name': 'Foo', 'building_height': 100, 'building_width': 100, 'building_length': 100, 'building_parapet_height': 100, 'wind_speed': 100, 'exposure_category': 'B', 'exposure_category_transition_distance': 5, 'ballast_block_weight': 12, 'max_system_pressure': 12, 'system_type': SystemType.dualTilt.value, 'module_type': ModuleType.Cell128.value, 'anchor_type': AnchorType.OMG_PowerGrip_Plus.value, 'design_spectral_response': 2, 'importance_factor': 1, } session_manager.save_form_submission(form_data) eq_(self.test_db_session.query(Site).count(), 1) site = self.test_db_session.query(Site).first() eq_(site.project_name, 'Foo') eq_(site.building_height, 100) eq_(site.building_width, 100) eq_(site.building_length, 100) eq_(site.parapet_height, 100) eq_(site.wind_speed, 100) eq_(site.exposure_category, 'B') eq_(site.exposure_transition_distance, 5) eq_(site.ballast_block_weight, 12) eq_(site.max_psf, 12) eq_(site.system_type, SystemType.dualTilt.value) eq_(site.module_type, ModuleType.Cell128.value) eq_(site.anchor_type, AnchorType.OMG_PowerGrip_Plus.value) eq_(site.spectral_response, 2) eq_(site.seismic_importance_factor, 1) def test_save_form_submission_updates_existing_project_site_if_already_exists_for_user(self): first_site = self.test_db_session.query(Site).first() first_site.cad_file = "hello world" first_site.dxf_file = "goodbye world" first_site.cad_file_name = "goodbye world" first_site.dxf_file_name = "goodbye world" eq_(self.test_db_session.query(Site).count(), 1) session_manager = SessionManager(self.session, None, self.test_db_session) form_data = { 'project_name': 'New Name', 'building_height': 105, 'building_width': 105, 'building_length': 105, 'building_parapet_height': 10, 'wind_speed': 110, 'exposure_category': 'B_C', 'exposure_category_transition_distance': 15, 'ballast_block_weight': 14, 'max_system_pressure': 20, 'system_type': SystemType.singleTilt.value, 'module_type': ModuleType.Cell96.value, 'anchor_type': AnchorType.OMG_PowerGrip.value, 'design_spectral_response': 1.5, 'importance_factor': 1.5, } session_manager.save_form_submission(form_data) eq_(self.test_db_session.query(Site).count(), 1) site = self.test_db_session.query(Site).first() eq_(site.project_name, 'New Name') eq_(site.building_height, 105) eq_(site.building_width, 105) eq_(site.building_length, 105) eq_(site.parapet_height, 10) eq_(site.wind_speed, 110) eq_(site.exposure_category, 'B_C') eq_(site.exposure_transition_distance, 15) eq_(site.ballast_block_weight, 14) eq_(site.max_psf, 20) eq_(site.system_type, SystemType.singleTilt.value) eq_(site.module_type, ModuleType.Cell96.value) eq_(site.anchor_type, AnchorType.OMG_PowerGrip.value) eq_(site.spectral_response, 1.5) eq_(site.seismic_importance_factor, 1.5) eq_(site.cad_file, None) eq_(site.dxf_file, None) eq_(site.cad_file_name, None) eq_(site.dxf_file_name, None) def test_save_form_submission_does_not_delete_cad_file_for_existing_site_if_system_or_module_type_did_not_change(self): first_site = self.test_db_session.query(Site).first() first_site.cad_file = "hello world" first_site.dxf_file = "goodbye world" first_site.cad_file_name = "goodbye world" first_site.dxf_file_name = "goodbye world" eq_(self.test_db_session.query(Site).count(), 1) session_manager = SessionManager(self.session, None, self.test_db_session) form_data = { 'project_name': 'New Name', 'building_height': 105, 'building_width': 105, 'building_length': 105, 'building_parapet_height': 10, 'wind_speed': 110, 'exposure_category': 'B_C', 'exposure_category_transition_distance': 15, 'ballast_block_weight': 14, 'max_system_pressure': 20, 'system_type': first_site.system_type, 'module_type': first_site.module_type, 'anchor_type': AnchorType.OMG_PowerGrip.value, 'design_spectral_response': 1.5, 'importance_factor': 1.5, } session_manager.save_form_submission(form_data) eq_(self.test_db_session.query(Site).count(), 1) site = self.test_db_session.query(Site).first() eq_(site.cad_file, "hello world") eq_(site.dxf_file, "goodbye world") eq_(site.cad_file_name, "goodbye world") eq_(site.dxf_file_name, "goodbye world") def test_save_uploaded_file(self): session_manager = SessionManager(self.session, self.redis_store, self.test_db_session) session_manager.save_uploaded_file("hello", cad_file_name="world", dxf_file_name="goodbye") site = self.test_db_session.query(Site).first() eq_(site.cad_file, "hello") eq_(site.cad_file_name, "world") eq_(site.dxf_file_name, "goodbye") def test_creating_power_stations_and_inverters(self): eq_(self.test_db_session.query(PowerStation).count(), 0) session_manager = SessionManager(self.session, None, self.test_db_session) power_station_form_data = { 'power_station_id': '', 'power_station_description': 'Test Power Station', 'power_station_quantity': 2, 'ac_run_length': 9, 'monitor_id': '', 'inverter_quantity': 1, 'inverter_1-model': 8, 'inverter_1-strings_per_inverter': 8, 'inverter_1-dc_switch': True, } session_manager.save_or_update_power_station(power_station_form_data) eq_(self.test_db_session.query(PowerStation).count(), 1) power_station = self.test_db_session.query(PowerStation).first() expected_power_station_data = { 'power_station_id': power_station.id, 'power_station_description': 'Test Power Station', 'power_station_quantity': 2, 'ac_run_length': 9, 'inverter_quantity': 1, 'inverters': [{ 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': False, 'dc_switch': True, 'splice_box': None, }] } assert_array_equal(session_manager.context()['power_stations'], [ expected_power_station_data ]) def test_updating_power_stations_and_inverters(self): power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() inverter = Inverter( model=str(InverterType.SMA.MODEL_24KW.value), strings_per_inverter=8, sunshade=False, dc_switch=True, power_station_id=power_station.id, standalone_inverter_id=None, ) self.test_db_session.add(inverter) self.test_db_session.commit() eq_(self.test_db_session.query(PowerStation).count(), 1) eq_(self.test_db_session.query(Inverter).count(), 1) session_manager = SessionManager(self.session, None, self.test_db_session) expected_power_station_data = { 'power_station_id': power_station.id, 'power_station_description': 'Test Power Station', 'power_station_quantity': 2, 'ac_run_length': 9, 'inverter_quantity': 1, 'inverters': [{ 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': False, 'dc_switch': True, 'splice_box': None, }] } assert_array_equal(session_manager.context()['power_stations'], [ expected_power_station_data ]) updated_power_station_data = { 'power_station_id': power_station.id, 'power_station_description': 'New Description', 'power_station_quantity': 4, 'ac_run_length': 10, 'inverter_quantity': 1, 'inverter_1-model': 4, 'inverter_1-strings_per_inverter': 3, 'inverter_1-dc_switch': True, 'inverter_1-sunshade': True, } session_manager.save_or_update_power_station(updated_power_station_data) assert_array_equal(session_manager.context()['power_stations'], [{ 'power_station_id': power_station.id, 'power_station_description': 'New Description', 'power_station_quantity': 4, 'ac_run_length': 10, 'inverter_quantity': 1, 'inverters': [{ 'model': InverterType.SMA.MODEL_12KW, 'strings_per_inverter': 3, 'sunshade': True, 'dc_switch': True, 'splice_box': None, }] }]) eq_(self.test_db_session.query(PowerStation).count(), 1) eq_(self.test_db_session.query(Inverter).count(), 1) def test_delete_power_station(self): power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() inverter = Inverter( model=str(InverterType.SMA.MODEL_24KW.value), strings_per_inverter=8, sunshade=False, dc_switch=True, power_station_id=power_station.id, standalone_inverter_id=None, ) self.test_db_session.add(inverter) self.test_db_session.commit() eq_(self.test_db_session.query(PowerStation).count(), 1) eq_(self.test_db_session.query(Inverter).count(), 1) session_manager = SessionManager(self.session, None, self.test_db_session) session_manager.delete_power_station(power_station.id) eq_(session_manager.context()['power_stations'], []) eq_(self.test_db_session.query(PowerStation).count(), 0) eq_(self.test_db_session.query(Inverter).count(), 0) def test_adding_power_monitor_to_switch_gear(self): eq_(len(self.test_db_session.query(PowerMonitor).all()), 0) session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['power_monitors'], []) monitor_form_data = { 'power_source': 'switch_gear', } session_manager.save_or_update_supervisor_monitor(monitor_form_data) eq_(len(self.test_db_session.query(PowerMonitor).all()), 1) power_monitor = self.test_db_session.query(PowerMonitor).first() eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Switch Gear/External', None) }]) def test_adding_power_monitor_to_power_station(self): session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['power_monitors'], []) power_monitors = self.test_db_session.query(PowerMonitor).all() eq_(len(power_monitors), 0) power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() monitor_form_data = { 'power_source': power_station.id, } session_manager.save_or_update_supervisor_monitor(monitor_form_data) eq_(len(self.test_db_session.query(PowerMonitor).all()), 1) power_monitor = self.test_db_session.query(PowerMonitor).first() eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Test Power Station', power_station.id) }]) def test_moving_power_monitor_from_switch_gear_to_power_station(self): power_monitor = PowerMonitor( site_id=self.site.id, power_station_id=None ) self.test_db_session.add(power_monitor) self.test_db_session.commit() power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() eq_(len(self.test_db_session.query(PowerMonitor).all()), 1) session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Switch Gear/External', None) }]) monitor_form_data = { 'monitor_id': power_monitor.id, 'power_source': power_station.id, } session_manager.save_or_update_supervisor_monitor(monitor_form_data) eq_(len(self.test_db_session.query(PowerMonitor).all()), 1) eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Test Power Station', power_station.id) }]) def test_moving_power_monitor_from_power_station_to_switch_gear(self): power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() power_monitor = PowerMonitor( site_id=self.site.id, power_station_id=power_station.id ) self.test_db_session.add(power_monitor) self.test_db_session.commit() eq_(len(self.test_db_session.query(PowerMonitor).all()), 1) session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Test Power Station', power_station.id) }]) monitor_form_data = { 'monitor_id': power_monitor.id, 'power_source': 'switch_gear' } session_manager.save_or_update_supervisor_monitor(monitor_form_data) eq_(len(self.test_db_session.query(PowerMonitor).all()), 1) eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Switch Gear/External', None) }]) def test_deleting_power_station_with_power_monitor_moves_monitor_to_switch_gear(self): power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() power_monitor = PowerMonitor( site_id=self.site.id, power_station_id=power_station.id ) self.test_db_session.add(power_monitor) self.test_db_session.commit() session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Test Power Station', power_station.id) }]) eq_(self.test_db_session.query(PowerStation).count(), 1) eq_(self.test_db_session.query(PowerMonitor).count(), 1) eq_(self.test_db_session.query(PowerMonitor).get(power_monitor.id).power_station_id, power_station.id) session_manager.delete_power_station(power_station.id) eq_(len(self.test_db_session.query(PowerMonitor).all()), 1) eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Switch Gear/External', None) }]) eq_(self.test_db_session.query(PowerStation).count(), 0) eq_(self.test_db_session.query(PowerMonitor).count(), 1) eq_(self.test_db_session.query(PowerMonitor).get(power_monitor.id).power_station_id, None) def test_deleting_power_monitor_attached_to_power_station_doesnt_otherwise_affect_power_station(self): power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() power_monitor = PowerMonitor( site_id=self.site.id, power_station_id=power_station.id ) self.test_db_session.add(power_monitor) self.test_db_session.commit() session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Test Power Station', power_station.id) }]) eq_(self.test_db_session.query(PowerStation).count(), 1) eq_(self.test_db_session.query(PowerMonitor).count(), 1) eq_(self.test_db_session.query(PowerMonitor).get(power_monitor.id).power_station_id, power_station.id) session_manager.delete_supervisor_monitor(power_monitor.id) eq_(len(self.test_db_session.query(PowerMonitor).all()), 0) eq_(session_manager.context()['power_monitors'], []) eq_(session_manager.context()['power_stations'], [{ 'inverter_quantity': 0, 'inverters': [], 'power_station_quantity': 2, 'power_station_description': 'Test Power Station', 'ac_run_length': 9, 'power_station_id': power_station.id }]) eq_(self.test_db_session.query(PowerStation).count(), 1) eq_(self.test_db_session.query(PowerMonitor).count(), 0) def test_deleting_power_monitor_attached_to_switch_gear(self): power_monitor = PowerMonitor( site_id=self.site.id, power_station_id=None ) self.test_db_session.add(power_monitor) self.test_db_session.commit() session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['power_monitors'], [{ 'monitor_id': power_monitor.id, 'power_source': ('Switch Gear/External', None) }]) eq_(self.test_db_session.query(PowerMonitor).count(), 1) session_manager.delete_supervisor_monitor(power_monitor.id) eq_(self.test_db_session.query(PowerMonitor).count(), 0) eq_(session_manager.context()['power_monitors'], []) def test_creating_standalone_inverters_switch_gear(self): eq_(self.test_db_session.query(StandaloneInverter).count(), 0) session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['standalone_inverters'], []) inverter_form_data = { 'standalone_inverter_id': '', 'standalone_ac_run_length': 70, 'inverter-quantity': 1, 'inverter-model': 8, 'inverter-strings_per_inverter': 8, 'inverter-sunshade': True, 'inverter-dc_switch': False, 'attachment_point': 'switch_gear', } session_manager.save_or_update_standalone_inverter(inverter_form_data) eq_(self.test_db_session.query(StandaloneInverter).count(), 1) standalone_inverter = self.test_db_session.query(StandaloneInverter).first() eq_(session_manager.context()['standalone_inverters'], [ { 'standalone_inverter_id': standalone_inverter.id, 'ac_run_length': 70, 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': True, 'dc_switch': False, 'splice_box': False, 'attachment_point': ('Switch Gear', None), } ]) def test_creating_standalone_inverters_attachment_point(self): power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() inverter = Inverter( model=str(InverterType.SMA.MODEL_24KW.value), strings_per_inverter=8, sunshade=False, dc_switch=True, power_station_id=power_station.id, standalone_inverter_id=None, ) self.test_db_session.add(inverter) self.test_db_session.commit() eq_(self.test_db_session.query(StandaloneInverter).count(), 0) session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['standalone_inverters'], []) inverter_form_data = { 'standalone_inverter_id': '', 'standalone_ac_run_length': 70, 'inverter-quantity': 1, 'inverter-model': 8, 'inverter-strings_per_inverter': 8, 'inverter-sunshade': True, 'attachment_point': str(power_station.id), } session_manager.save_or_update_standalone_inverter(inverter_form_data) eq_(self.test_db_session.query(StandaloneInverter).count(), 1) standalone_inverter = self.test_db_session.query(StandaloneInverter).first() eq_(session_manager.context()['standalone_inverters'], [ { 'standalone_inverter_id': standalone_inverter.id, 'ac_run_length': 70, 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': True, 'dc_switch': False, 'splice_box': False, 'attachment_point': (power_station.description, power_station.id), } ]) def test_updating_standalone_inverters(self): standalone_inverter = StandaloneInverter( site_id=self.site.id, attachment_point_id=None, ac_run_length=70, ) self.test_db_session.add(standalone_inverter) self.test_db_session.commit() inverter = Inverter( model=str(InverterType.SMA.MODEL_24KW.value), strings_per_inverter=8, sunshade=True, dc_switch=False, power_station_id=None, standalone_inverter_id=standalone_inverter.id, ) self.test_db_session.add(inverter) self.test_db_session.commit() eq_(self.test_db_session.query(StandaloneInverter).count(), 1) session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['standalone_inverters'], [ { 'standalone_inverter_id': standalone_inverter.id, 'ac_run_length': 70, 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': True, 'dc_switch': False, 'splice_box': None, 'attachment_point': ('Switch Gear', None), } ]) power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() updated_standalone_inverter_data = { 'standalone_inverter_id': standalone_inverter.id, 'standalone_ac_run_length': 69, 'inverter-model': 6, 'inverter-strings_per_inverter': 7, 'inverter-sunshade': False, 'inverter-dc_switch': True, 'attachment_point': power_station.id, } session_manager.save_or_update_standalone_inverter(updated_standalone_inverter_data) eq_(self.test_db_session.query(StandaloneInverter).count(), 1) eq_(session_manager.context()['standalone_inverters'], [ { 'standalone_inverter_id': standalone_inverter.id, 'ac_run_length': 69, 'model': InverterType.SMA.MODEL_20KW, 'strings_per_inverter': 7, 'sunshade': False, 'dc_switch': True, 'splice_box': False, 'attachment_point': ('Test Power Station', power_station.id), } ]) def test_delete_standalone_inverters(self): standalone_inverter = StandaloneInverter( site_id=self.site.id, attachment_point_id=None, ac_run_length=70, ) self.test_db_session.add(standalone_inverter) self.test_db_session.commit() inverter = Inverter( model=str(InverterType.SMA.MODEL_24KW.value), strings_per_inverter=8, sunshade=True, dc_switch=False, power_station_id=None, standalone_inverter_id=standalone_inverter.id, ) self.test_db_session.add(inverter) self.test_db_session.commit() eq_(self.test_db_session.query(StandaloneInverter).count(), 1) eq_(self.test_db_session.query(Inverter).count(), 1) session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['standalone_inverters'], [ { 'standalone_inverter_id': standalone_inverter.id, 'ac_run_length': 70, 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': True, 'dc_switch': False, 'splice_box': None, 'attachment_point': ('Switch Gear', None), } ]) session_manager.delete_standalone_inverter(standalone_inverter.id) eq_(session_manager.context()['standalone_inverters'], []) eq_(self.test_db_session.query(StandaloneInverter).count(), 0) eq_(self.test_db_session.query(Inverter).count(), 0) def test_delete_power_station_with_attached_standalone_inverter_changes_inverter_to_switchgear(self): # power station with 1 inverter power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() inverter = Inverter( # attached to the above power station model=str(InverterType.SMA.MODEL_24KW.value), strings_per_inverter=8, sunshade=False, dc_switch=True, power_station_id=power_station.id, standalone_inverter_id=None, ) self.test_db_session.add(inverter) self.test_db_session.commit() # standalone inverter, attached to the power station standalone_inverter = StandaloneInverter( site_id=self.site.id, attachment_point_id=power_station.id, ac_run_length=70, ) self.test_db_session.add(standalone_inverter) self.test_db_session.commit() inverter_2 = Inverter( # attached to the above standalone inverter model=str(InverterType.SMA.MODEL_24KW.value), strings_per_inverter=8, sunshade=True, dc_switch=False, power_station_id=None, standalone_inverter_id=standalone_inverter.id, ) self.test_db_session.add(inverter_2) self.test_db_session.commit() session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['standalone_inverters'], [ { 'standalone_inverter_id': standalone_inverter.id, 'ac_run_length': 70, 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': True, 'dc_switch': False, 'splice_box': None, 'attachment_point': ('Test Power Station', power_station.id), } ]) eq_(self.test_db_session.query(PowerStation).count(), 1) eq_(self.test_db_session.query(Inverter).count(), 2) eq_(self.test_db_session.query(StandaloneInverter).count(), 1) eq_(self.test_db_session.query(StandaloneInverter).get(standalone_inverter.id).attachment_point_id, power_station.id) session_manager.delete_power_station(power_station.id) eq_(self.test_db_session.query(PowerStation).count(), 0) eq_(self.test_db_session.query(Inverter).count(), 1) eq_(self.test_db_session.query(StandaloneInverter).count(), 1) eq_(self.test_db_session.query(StandaloneInverter).get(standalone_inverter.id).attachment_point_id, None) eq_(session_manager.context()['power_stations'], []) eq_(session_manager.context()['standalone_inverters'], [ { 'standalone_inverter_id': standalone_inverter.id, 'ac_run_length': 70, 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': True, 'dc_switch': False, 'splice_box': None, 'attachment_point': ('Switch Gear', None), } ]) def test_deleting_standalone_inverter_attached_to_power_station_doesnt_otherwise_change_power_station(self): # power station with 1 inverter power_station = PowerStation( site_id=self.site.id, quantity=2, ac_run_length=9, description='Test Power Station', ) self.test_db_session.add(power_station) self.test_db_session.commit() inverter = Inverter( # attached to the above power station model=str(InverterType.SMA.MODEL_24KW.value), strings_per_inverter=8, sunshade=False, dc_switch=True, power_station_id=power_station.id, standalone_inverter_id=None, ) self.test_db_session.add(inverter) self.test_db_session.commit() # standalone inverter, attached to the power station standalone_inverter = StandaloneInverter( site_id=self.site.id, attachment_point_id=power_station.id, ac_run_length=70, ) self.test_db_session.add(standalone_inverter) self.test_db_session.commit() inverter_2 = Inverter( # attached to the above standalone inverter model=str(InverterType.SMA.MODEL_24KW.value), strings_per_inverter=8, sunshade=True, dc_switch=False, power_station_id=None, standalone_inverter_id=standalone_inverter.id, ) self.test_db_session.add(inverter_2) self.test_db_session.commit() eq_(self.test_db_session.query(PowerStation).count(), 1) eq_(self.test_db_session.query(Inverter).count(), 2) eq_(self.test_db_session.query(StandaloneInverter).count(), 1) session_manager = SessionManager(self.session, None, self.test_db_session) eq_(session_manager.context()['standalone_inverters'], [ { 'standalone_inverter_id': standalone_inverter.id, 'ac_run_length': 70, 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': True, 'dc_switch': False, 'splice_box': None, 'attachment_point': ('Test Power Station', power_station.id), } ]) session_manager.delete_standalone_inverter(standalone_inverter.id) eq_(self.test_db_session.query(PowerStation).count(), 1) eq_(self.test_db_session.query(Inverter).count(), 1) eq_(self.test_db_session.query(StandaloneInverter).count(), 0) eq_(session_manager.context()['standalone_inverters'], []) eq_(session_manager.context()['power_stations'], [{ 'power_station_id': power_station.id, 'power_station_description': 'Test Power Station', 'power_station_quantity': 2, 'ac_run_length': 9, 'inverter_quantity': 1, 'inverters': [{ 'model': InverterType.SMA.MODEL_24KW, 'strings_per_inverter': 8, 'sunshade': False, 'dc_switch': True, 'splice_box': None, }] }]) def test_uploading_csv_deletes_any_user_provided_seismic_anchors(self): session_manager = SessionManager(self.session, self.redis_store, self.test_db_session) session_manager.save_user_provided_seismic_anchors([{"seismic_anchors": 1}]) user_values = session_manager.user_values() eq_(user_values.user_override_seismic_anchors(), True) session_manager.save_uploaded_file("hello") eq_(user_values.user_override_seismic_anchors(), False)