(self, context: Context)
| 113 | return self._contexts[self._current_context] |
| 114 | |
| 115 | def set_and_write(self, context: Context): |
| 116 | # treat the same endpoint with same services as the same coordinator |
| 117 | for _, v in self._contexts.items(): |
| 118 | if ( |
| 119 | context.coordinator_endpoint == v.coordinator_endpoint |
| 120 | and context.flex == v.flex |
| 121 | ): |
| 122 | # reset to global context |
| 123 | v.switch_context("global") |
| 124 | contexts = [v.to_dict() for _, v in self._contexts.items()] |
| 125 | write_yaml_file( |
| 126 | {"contexts": contexts, "current-context": self._current_context}, |
| 127 | GS_CONFIG_DEFAULT_LOCATION, |
| 128 | ) |
| 129 | return |
| 130 | |
| 131 | # set |
| 132 | self._current_context = context.name |
| 133 | self._contexts[context.name] = context |
| 134 | |
| 135 | # write |
| 136 | contexts = [v.to_dict() for _, v in self._contexts.items()] |
| 137 | write_yaml_file( |
| 138 | {"contexts": contexts, "current-context": self._current_context}, |
| 139 | GS_CONFIG_DEFAULT_LOCATION, |
| 140 | ) |
| 141 | |
| 142 | def update_and_write(self, context: Context): |
| 143 | if context.name not in self._contexts: |
no test coverage detected