1 # ==================================================================================
2 # Copyright (c) 2019 Nokia
3 # Copyright (c) 2018-2019 AT&T Intellectual Property.
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 # ==================================================================================
19 from ricxappframe.entities.rnib.nb_identity_pb2 import NbIdentity
20 from ricxappframe.entities.rnib.cell_pb2 import Cell
21 import ricxappframe.entities.rnib.nb_identity_pb2 as pb_nb
22 import ricxappframe.entities.rnib.nodeb_info_pb2 as pb_nbi
23 import ricxappframe.entities.rnib.enb_pb2 as pb_enb
24 import ricxappframe.entities.rnib.gnb_pb2 as pb_gnb
25 import ricxappframe.entities.rnib.cell_pb2 as pb_cell
26 import ricxappframe.entities.rnib.ran_function_pb2 as pb_rf
29 # These are here just to reduce the size of the code in test_rmr so those (important) tests are more readable; in theory these dicts could be large
30 # The actual value of the constants should be ignored by the tests; all we should care
31 # about is that the constant value was returned by the RMR function. Further, we should
32 # not consider it an error if RMR returns more than what is listed here; these are the
33 # list of what is/could be used by this package.
35 def expected_constants():
41 "RMR_MAX_RCV_BYTES": 4096,
43 "RMRFL_MTCALL": 2, # can't be added here until jenkins version >= 1.8.3
44 "RMRFL_AUTO_ALLOC": 3,
46 "RMR_VOID_MSGTYPE": -1,
53 "RMR_ERR_SENDFAILED": 5,
54 "RMR_ERR_CALLFAILED": 6,
55 "RMR_ERR_NOWHOPEN": 7,
57 "RMR_ERR_OVERFLOW": 9,
59 "RMR_ERR_RCVFAILED": 11,
60 "RMR_ERR_TIMEOUT": 12,
63 "RMR_ERR_INITFAILED": 15,
68 def expected_states():
75 5: "RMR_ERR_SENDFAILED",
76 6: "RMR_ERR_CALLFAILED",
77 7: "RMR_ERR_NOWHOPEN",
79 9: "RMR_ERR_OVERFLOW",
81 11: "RMR_ERR_RCVFAILED",
82 12: "RMR_ERR_TIMEOUT",
85 15: "RMR_ERR_INITFAILED",
90 def rnib_information():
92 rnib1.inventory_name = "nodeb_1234"
93 rnib1.global_nb_id.plmn_id = "plmn_1234"
94 rnib1.global_nb_id.nb_id = "nb_1234"
95 rnib1.connection_status = 1
98 rnib1.inventory_name = "nodeb_5678"
99 rnib1.global_nb_id.plmn_id = "plmn_5678"
100 rnib1.global_nb_id.nb_id = "nb_5678"
101 rnib1.connection_status = 6
103 return [rnib1.SerializeToString(), rnib2.SerializeToString()]
107 def rnib_cellinformation():
110 rnib_cell1.type = Cell.LTE_CELL
113 rnib_cell2.type = Cell.LTE_CELL
115 return [rnib_cell2.SerializeToString(), rnib_cell2.SerializeToString()]
118 def createRanFunction(ran_function_id, definition, revision, oid):
119 r = pb_rf.RanFunction()
120 r.ran_function_id = ran_function_id
121 r.ran_function_definition = definition
122 r.ran_function_revision = revision
123 r.ran_function_oid = oid
127 def createServedCellInfo(pci, cell_id, tac):
128 s = pb_enb.ServedCellInfo()
135 def createServedNRCellInfo(pci, cell_id, tac):
136 s = pb_gnb.ServedNRCell()
137 # s = pb_gnb.ServedNRCellInformation()
138 s.served_nr_cell_information.nr_pci = pci
139 s.served_nr_cell_information.cell_id = cell_id
140 s.served_nr_cell_information.stac5g = tac
141 s.served_nr_cell_information.nr_mode = 2
147 def createNodeb(name, plmn, nbid):
148 nb = pb_nb.NbIdentity()
149 nb.inventory_name = name
150 nb.global_nb_id.plmn_id = plmn
151 nb.global_nb_id.nb_id = nbid
152 nb.connection_status = 1
156 def createNodebInfo(name, nbtype, ip, port):
157 nbi = pb_nbi.NodebInfo()
160 nbi.e2_application_protocol = 1
161 nbi.connection_status = 1
164 rf1 = createRanFunction(1, b'te524367153', 1, "1.3.6.1.4.1.1.2.2.1")
165 rf2 = createRanFunction(1, b'te524367153', 1, "1.3.6.1.4.1.1.2.2.2")
166 rf3 = createRanFunction(1, b'te524367153', 1, "1.3.6.1.4.1.1.2.2.3")
167 nbi.gnb.ran_functions.extend([rf1, rf2, rf3])
168 s1 = createServedNRCellInfo(1, b'822', b'1452')
169 s2 = createServedNRCellInfo(2, b'823', b'1453')
170 s3 = createServedNRCellInfo(3, b'824', b'1454')
171 nbi.gnb.served_nr_cells.extend([s1, s2, s3])
175 s1 = createServedCellInfo(1, b'822', b'1452')
176 s2 = createServedCellInfo(2, b'823', b'1453')
177 s3 = createServedCellInfo(3, b'824', b'1454')
178 nbi.enb.served_cells.extend([s1, s2, s3])
182 def createCell(name, pci):
184 c.type = pb_cell.Cell.Type.Value('LTE_CELL')
185 # print(pb_cell.Cell.Type.Name(pb_cell.Cell.LTE_CELL))
186 c.served_cell_info.pci = pci
187 c.served_cell_info.cell_id = b'822'
188 c.served_cell_info.tac = b'1452'