diff --git a/medcat-v2/README.md b/medcat-v2/README.md index a8248c195..244b1a4c3 100644 --- a/medcat-v2/README.md +++ b/medcat-v2/README.md @@ -151,6 +151,29 @@ See [Architecture Documentation](docs/architecture.md) for detailed information. ## Tutorials A guide on how to use MedCAT v2 is available at on the medcat documentation page on [docs.cogstack.org](https://docs.cogstack.org) +## Contributing + +Please follow the [Contribution Guidelines](../CONTRIBUTING.md). + +When writing your own component (NER or linker), it is recommended making sure they follow the contracts for these components. +
+Example test for custom components in model pack + +```python +from unittest import TestCase +from medcat.components.contracting_testing import assert_component_contracts +# implement create_model_with_my_component +class MyComponentTest(TestCase): + def test_my_model_contract(self): + # create or load a model with your custom component(s) + # NOTE: This would (generally) need to be able to NER / link 1 entity in the example text + # The test time models in medcat would be sufficient + cat = create_model_with_my_component() + assert_component_contracts(cat) +``` +
+ + ## Acknowledgements Entity extraction was trained on [MedMentions](https://github.com/chanzuckerberg/MedMentions) In total it has ~ 35K entites from UMLS diff --git a/medcat-v2/medcat/components/base.py b/medcat-v2/medcat/components/base.py new file mode 100644 index 000000000..2d64b945c --- /dev/null +++ b/medcat-v2/medcat/components/base.py @@ -0,0 +1,109 @@ +from typing import Protocol, runtime_checkable, Optional +from typing_extensions import Self +from enum import Enum + +from pydantic import BaseModel + +from medcat.tokenizing.tokens import MutableDocument +from medcat.tokenizing.tokenizers import BaseTokenizer +from medcat.cdb import CDB +from medcat.vocab import Vocab +from medcat.config.config import ComponentConfig + + +@runtime_checkable +class BaseComponent(Protocol): + + @property + def full_name(self) -> Optional[str]: + """Name with the component type (e.g ner, linking, meta).""" + pass + + @property + def name(self) -> str: + """The name of the component.""" + pass + + def is_core(self) -> bool: + """Whether the component is a core component or not. + + Returns: + bool: Whether this is a core component. + """ + pass + + def __call__(self, doc: MutableDocument) -> MutableDocument: + pass + + @classmethod + def create_new_component( + cls, cnf: ComponentConfig, tokenizer: BaseTokenizer, + cdb: CDB, vocab: Vocab, model_load_path: Optional[str]) -> Self: + """Create a new component or load one off disk if load path presented. + + This may raise an exception if the wrong type of config is provided. + + Args: + cnf (ComponentConfig): The config relevant to this components. + tokenizer (BaseTokenizer): The base tokenizer. + cdb (CDB): The CDB. + vocab (Vocab): The Vocab. + model_load_path (Optional[str]): Model load path (if present). + + Returns: + Self: The new components. + """ + pass + +class CollectionContract(BaseModel, frozen=True): + """Contract for a collection field — what each item in the collection provides.""" + field: str # e.g. 'ner_ents' + must_provide: frozenset[str] # fields every item must have + may_provide: frozenset[str] = frozenset() + + +class ComponentContract(BaseModel, frozen=True): + needs: frozenset[str] + must_provide: frozenset[str] + may_provide: frozenset[str] = frozenset() + collection_contracts: frozenset[CollectionContract] = frozenset() + + +class CoreComponentType(Enum): + tagging = ComponentContract( + needs=frozenset(), + must_provide=frozenset(), + # doesn't write for every token + may_provide=frozenset({'token.is_punctuation', 'token.to_skip'}), + collection_contracts=frozenset(), + ) + token_normalizing = ComponentContract( + needs=frozenset(), + # should write for every token + must_provide=frozenset({'token.norm'}), + may_provide=frozenset(), + collection_contracts=frozenset(), + ) + ner = ComponentContract( + needs=frozenset({'token.to_skip'}), + must_provide=frozenset({'doc.ner_ents'}), # the list must exist + may_provide=frozenset(), + collection_contracts=frozenset({ + CollectionContract( + field='doc.ner_ents', + must_provide=frozenset({'detected_name'}), + ) + }), + ) + linking = ComponentContract( + needs=frozenset({'doc.ner_ents'}), + # must write, but may be empty list + must_provide=frozenset({'doc.linked_ents'}), + may_provide=frozenset({}), + collection_contracts=frozenset({ + CollectionContract( + field='doc.linked_ents', + must_provide=frozenset({'cui', 'context_similarity'}), + ), + }), + ) diff --git a/medcat-v2/medcat/components/contracting.py b/medcat-v2/medcat/components/contracting.py new file mode 100644 index 000000000..8afe55e40 --- /dev/null +++ b/medcat-v2/medcat/components/contracting.py @@ -0,0 +1,241 @@ +from typing import Callable +from logging import Logger + +from medcat.tokenizing.tokens import MutableDocument +from medcat.components.base import BaseComponent, ComponentContract +from medcat.components.contracting_utils import ( + AccessType, wrap_relevant_parts, ContractViolation) + + +logger = Logger(__name__) + + +def verify_part( + text: str, + doc_getter: Callable[[str], MutableDocument], + component: BaseComponent, + paths: list[str], + access_type: AccessType, + raise_on_violation: bool = True, + min_feedbacks: int = 0, +) -> list[str]: + """Verify the parts for this text. + + Args: + text (str): The text to use. + doc_getter (Callable[[str], MutableDocument]): The document getter. + component (BaseComponent): The component to check. + paths (list[str]): The paths to check. + access_type (AccessType): The type of access to check. + raise_on_violation (bool): Whether to raise on a violation. + Defaults to True. + min_feedbacks (int): The minimum number of feedbacks expected. + Defaults to 0. + + Raises: + ContractViolation: If there are violations and instructed to raise. + + Returns: + list[str]: The list of violations, if any. + """ + violations: list[str] = [] + for path in paths: + doc = doc_getter(text) + with wrap_relevant_parts(doc, path) as feedback: + doc = component(doc) + # verify each one access + accessed = sum(bool(fb) for fb in feedback) + total = len(feedback) + if accessed != total: + violations.append( + f"Component {component.full_name} does not {access_type.name} " + f"{path} ({accessed} / {total} accessed)") + logger.debug( + "Found a virolation in component '%s' for %s at '%s': " + "(%d / %d) accessed with feedback %s", + component.full_name, access_type.name, + path, accessed, total, feedback, + ) + elif total < min_feedbacks: + violations.append( + f"Component {component.full_name} did not {access_type.name} " + f"{path} enough ({total} with minimum {min_feedbacks})") + logger.debug( + "Component '%s' did not %s " + "'%s' enough (%d with minimum %d)", + component.full_name, access_type.name, + path, total, min_feedbacks, + ) + if raise_on_violation: + raise ContractViolation("\n".join(violations)) + return violations + + +def verify_needs( + text: str, + doc_getter: Callable[[str], MutableDocument], + component: BaseComponent, + contract: ComponentContract, + raise_on_violation: bool = True, + min_feedbacks: int = 0, +) -> list[str]: + """Verify the needs portion of a contract. + + Args: + text (str): The text to use. + doc_getter (Callable[[str], MutableDocument]): The document getter. + component (BaseComponent): The component to check. + contract (ComponentContract): The contract to check. + raise_on_violation (bool): Whether to raise on violations. + Defaults to True. + min_feedbacks (int,): The minimum number of feedbacks expected. + Defaults to 0. + + Returns: + list[str]: The list of violations, if any. + """ + return verify_part( + text, doc_getter, component, list(contract.needs), + AccessType.READ, raise_on_violation=raise_on_violation, + min_feedbacks=min_feedbacks, + ) + + +def verify_must_provide( + text: str, + doc_getter: Callable[[str], MutableDocument], + component: BaseComponent, + contract: ComponentContract, + raise_on_violation: bool = True, + min_feedbacks: int = 0, +) -> list[str]: + """Verify the must-provide portion of a contract. + + Args: + text (str): The text to use. + doc_getter (Callable[[str], MutableDocument]): The document getter. + component (BaseComponent): The component to check. + contract (ComponentContract): The contract to check. + raise_on_violation (bool): Whether to raise on violations. + Defaults to True. + min_feedbacks (int,): The minimum number of feedbacks expected. + Defaults to 0. + + Returns: + list[str]: The list of violations, if any. + """ + return verify_part( + text, doc_getter, component, list(contract.must_provide), + AccessType.WRITE, raise_on_violation=raise_on_violation, + min_feedbacks=min_feedbacks, + ) + + +def verify_collections_contracts( + text: str, + doc_getter: Callable[[str], MutableDocument], + component: BaseComponent, + contract: ComponentContract, + raise_on_violation: bool = True, + min_feedbacks: int = 0, +) -> list[str]: + """Verify the collection contracts portion of a contract. + + This method only really checks the length of the collection + and that each item in there has a truthy value for each required + field. The expectation is that the write action (for the collection) + is checked by other parts. And because these entities may be created + in order to put them in the lists (i.e for NER) without the data filled + in, it's fair to assume that if the data exists, it was filled in. + + Args: + text (str): The text to use. + doc_getter (Callable[[str], MutableDocument]): The document getter. + component (BaseComponent): The component to check. + contract (ComponentContract): The contract to check. + raise_on_violation (bool): Whether to raise on violations. + Defaults to True. + min_feedbacks (int): The minimum number of feedbacks expected. + Defaults to 0. + + Returns: + list[str]: The list of violations, if any. + """ + violations: list[str] = [] + if not contract.collection_contracts: + return violations + doc = doc_getter(text) + doc = component(doc) + for cc in contract.collection_contracts: + if not cc.field.startswith("doc."): + violations.append( + f"Collection contract field '{cc.field}' is not a doc-level " + f"field — only doc.* fields are currently supported") + continue + attr = cc.field.split(".", 1)[1] + try: + collection = getattr(doc, attr) + except AttributeError: + violations.append( + f"Component {component.full_name} did not provide " + f"collection '{cc.field}' at all") + continue + items = list(collection) + if len(items) < min_feedbacks: + violations.append( + f"Collection '{cc.field}' has too few items " + f"({len(items)} with minimum {min_feedbacks})") + for i, item in enumerate(items): + for field in cc.must_provide: + try: + val = getattr(item, field) + except AttributeError: + violations.append( + f"Item {i} in '{cc.field}' is missing " + f"required field '{field}'") + continue + if not val: + violations.append( + f"Item {i} in '{cc.field}' has falsy value " + f"for required field '{field}' (got {val!r})") + if violations and raise_on_violation: + raise ContractViolation("\n".join(violations)) + return violations + + +def verify_contract( + text: str, + doc_getter: Callable[[str], MutableDocument], + component: BaseComponent, + contract: ComponentContract, + raise_on_violation: bool = True, + min_feedbacks_need: int = 0, + min_feedbacks_provide: int = 0, + min_feedbacks_contracts: int = 0, +) -> list[str]: + """ + Verify a ComponentContract against a document before/after a component ran. + Returns a list of violation messages. + + Raises ContractViolation if violations found and raise_on_violation. + """ + # verify needs are met + violations = verify_needs( + text, doc_getter, component, contract, raise_on_violation=False, + min_feedbacks=min_feedbacks_need, + ) + # verify mandatory returns are done + violations += verify_must_provide( + text, doc_getter, component, contract, raise_on_violation=False, + min_feedbacks=min_feedbacks_provide, + ) + # verify collections contracts + violations += verify_collections_contracts( + text, doc_getter, component, contract, raise_on_violation=False, + min_feedbacks=min_feedbacks_contracts, + ) + + if violations and raise_on_violation: + raise ContractViolation("\n".join(violations)) + + return violations diff --git a/medcat-v2/medcat/components/contracting_testing.py b/medcat-v2/medcat/components/contracting_testing.py new file mode 100644 index 000000000..9cfffd75d --- /dev/null +++ b/medcat-v2/medcat/components/contracting_testing.py @@ -0,0 +1,109 @@ +from typing import Optional + +from medcat.cat import CAT +from medcat.tokenizing.tokens import MutableDocument +from medcat.components.base import CoreComponentType +from medcat.components.types import CoreComponent +from medcat.components.contracting import verify_contract + + +_DEFAULT_CONTRACT_TEXT = """ +John had been diagnosed with acute Kidney - Failure the week before. +""" +_DEFAULT_COMP_TYPES_TO_CHECK = [ + CoreComponentType.ner, CoreComponentType.linking] + + +class ContractViolationError(ValueError): + + def __init__(self, component_type: CoreComponentType, violations: list): + self.component_type = component_type + self.violations = violations + super().__init__( + f"Contract violations for {component_type.name}: {violations}" + ) + + +def assert_single_component_holds( + model: CAT, + component: CoreComponent, + text: str = _DEFAULT_CONTRACT_TEXT, +): + """Assert a specific component's contract holds. + + Example: + + def test_my_ner_contract(self): + cat = create_model_with_my_ner() + my_ner = cat.pipe.get_component(CoreComponentType.ner) + assert_single_component_holds(cat, my_ner) + + Args: + model (CAT): The model with the specific component. + component (CoreComponent): The component under test. + text (str): The text to use for the check. + Defaults to _DEFAULT_CONTRACT_TEXT. + """ + component_type = component.get_type() + + def prep(t: str) -> MutableDocument: + return model.pipe.pipe_until(t, component_type) + + contract = component_type.value + min_feedbacks_need = ( + len(list(prep(text))) if component_type is CoreComponentType.ner else 1 + ) + if not min_feedbacks_need: + # NOTE: this would normally happen with NER if/when there's no tokens + raise ContractViolationError( + component_type, + ["Cannot check for feedback needs if minimum is 0 " + f"for {component.full_name}", ]) + violations = verify_contract( + text, prep, component, contract, + raise_on_violation=False, + min_feedbacks_need=min_feedbacks_need, + # NOTE: this means that the collection (ner_ents or linked_ents) is + # written to at least once + min_feedbacks_provide=1, + # NOTE: this means we expect at least 1 entity in output + min_feedbacks_contracts=1, + ) + if violations: + raise ContractViolationError(component_type, violations) + + +def assert_component_contracts( + model: CAT, + text: str = _DEFAULT_CONTRACT_TEXT, + to_check: Optional[list[CoreComponentType]] = None +): + """Verify that all components upholds its MedCAT contract. + + Intended for use in tests by external component implementers. + Raises ContractViolationError if the contract is not upheld. + + Example: + + def test_my_model_contract(self): + cat = create_model_with_my_component() + assert_component_contracts(cat) + + Args: + model (CAT): The model pack to use. This needs to refer to a model that + is able to NER and link at least 1 entity in the provided text. + This model needs to already have the relevant component(s) to be + checked. + to_check (Optional[list[CoreComponentType]]): The core component types + to check. Defaults to NER and linking. + text (str): The text to use for the check. + Defaults to _DEFAULT_CONTRACT_TEXT. + + Raises: + ContractViolationError: If there are any violations found. + """ + if to_check is None: + to_check = _DEFAULT_COMP_TYPES_TO_CHECK + for cct in to_check: + cur_comp = model.pipe.get_component(cct) + assert_single_component_holds(model, cur_comp, text) diff --git a/medcat-v2/medcat/components/contracting_utils.py b/medcat-v2/medcat/components/contracting_utils.py new file mode 100644 index 000000000..cde9ed741 --- /dev/null +++ b/medcat-v2/medcat/components/contracting_utils.py @@ -0,0 +1,183 @@ +from typing import Any, Iterator, Type +from contextlib import contextmanager, ExitStack +from collections import defaultdict +from logging import Logger +from enum import Enum, auto + +from medcat.tokenizing.tokens import MutableDocument + + +logger = Logger(__name__) + +_SENTINEL = object() + + +class ContractViolation(Exception): + pass + + +class AccessType(Enum): + READ = auto() + WRITE = auto() + + +def iter_relevant_parts(doc: MutableDocument, path: str) -> Iterator[Any]: + if path.startswith("doc."): + yield doc + return + if path.startswith("token."): + yield from doc[:] + else: + raise ValueError(f"Unknown path: {path}") + + +class WrappedMember: + + def __init__( + self, + part: Any, + member_name: str, + feedback: list[Any], + access_type: AccessType, + ) -> None: + self.part = part + self.member_name = member_name + self.feedback = feedback + self.access_type = access_type + self._oirg_class = type(self.part) + self._install() + + def _install(self): + original_cls = type(self.part) + spy = self # capture for closure + + if self.access_type == AccessType.READ: + + class SpySubclass(original_cls): + def __getattribute__(self, name): + val = super().__getattribute__(name) + if name == spy.member_name: + # saving str copy of value + spy.feedback.append(str(val)) + return val + + elif self.access_type == AccessType.WRITE: + + class SpySubclass(original_cls): + def __setattr__(self, name, value): + if name in spy.member_name: + try: + old = super().__getattribute__(name) + except AttributeError: + old = AttributeError # sentinel: didn't exist yet + # saving str copies of state + spy.feedback.append((str(old), str(value))) + super().__setattr__(name, value) + + SpySubclass.__name__ = f"Spy({original_cls.__name__})" + SpySubclass.__qualname__ = SpySubclass.__name__ + self.part.__class__ = SpySubclass + + def __enter__(self): + return self + + def __exit__(self, *_): + self.part.__class__ = self._oirg_class + + +@contextmanager +def spy_token_class( + token_cls: Type, + watched_attr: str, + access_type: AccessType, +): + prev_getattr = token_cls.__dict__.get('__getattribute__', _SENTINEL) + prev_setattr = token_cls.__dict__.get('__setattr__', _SENTINEL) + + per_instance_spied: dict[Any, list[Any]] = defaultdict(list) + + def __getattribute__(self, name: str) -> Any: + val = object.__getattribute__(self, name) + if name == watched_attr: + per_instance_spied[self].append(str(val)) + return val + + def __setattr__(self, name: str, value: Any): + old = object.__getattribute__(self, name) + object.__setattr__(self, name, value) + if name == watched_attr: + per_instance_spied[self].append((str(old), str(value))) + + if access_type == AccessType.READ: + # NOTE: this should be fine, but mypy complains due to self + token_cls.__getattribute__ = __getattribute__ # type: ignore + elif access_type == AccessType.WRITE: + # NOTE: this should be fine, but mypy complains due to self + token_cls.__setattr__ = __setattr__ # type: ignore + else: + raise ValueError(f"Unknown access type: {access_type}") + try: + yield per_instance_spied + finally: + if prev_getattr is _SENTINEL: + del token_cls.__getattribute__ + else: + token_cls.__getattribute__ = prev_getattr + + if prev_setattr is _SENTINEL: + # NOTE: this should be fine, but mypy complains due to self + token_cls.__setattr__ = object.__setattr__ # type: ignore + else: + token_cls.__setattr__ = prev_setattr + + +@contextmanager +def wrap_relevant_parts( + doc: MutableDocument, + path: str, + access_type: AccessType = AccessType.READ, +): + if path.startswith("doc."): + with wrap_relevant_persistant_parts( + doc, path, access_type + ) as feedbacks: + yield feedbacks + elif path.startswith("token."): + with wrap_relevant_token_cls( + doc, path, access_type + ) as feedbacks: + yield feedbacks + + +@contextmanager +def wrap_relevant_token_cls( + doc: MutableDocument, + path: str, + access_type: AccessType = AccessType.READ, +): + _, attr_name = path.split(".", 1) + tkn_cls = type(next(iter(doc))) + with spy_token_class( + tkn_cls, attr_name, access_type + ) as per_instance_spied: + yield per_instance_spied.values() + + +@contextmanager +def wrap_relevant_persistant_parts( + doc: MutableDocument, + path: str, + access_type: AccessType = AccessType.READ, +): + member_name = path.split(".", 1)[1] + out_list: list[list[Any]] = [] + with ExitStack() as exit_stack: + for part in iter_relevant_parts(doc, path): + feedback: list[Any] = [] + exit_stack.enter_context( + WrappedMember( + part, member_name, + feedback, access_type=access_type) + ) + out_list.append(feedback) + yield out_list diff --git a/medcat-v2/medcat/components/types.py b/medcat-v2/medcat/components/types.py index fe26bb1e0..76e1ede74 100644 --- a/medcat-v2/medcat/components/types.py +++ b/medcat-v2/medcat/components/types.py @@ -1,7 +1,5 @@ from typing import Optional, Protocol, Callable, runtime_checkable, Union from typing import Literal -from typing_extensions import Self -from enum import Enum, auto from abc import ABC, abstractmethod from medcat.utils.registry import Registry, MedCATRegistryException @@ -10,58 +8,7 @@ from medcat.cdb import CDB from medcat.vocab import Vocab from medcat.config.config import ComponentConfig - - -class CoreComponentType(Enum): - tagging = auto() - token_normalizing = auto() - ner = auto() - linking = auto() - - -@runtime_checkable -class BaseComponent(Protocol): - - @property - def full_name(self) -> Optional[str]: - """Name with the component type (e.g ner, linking, meta).""" - pass - - @property - def name(self) -> str: - """The name of the component.""" - pass - - def is_core(self) -> bool: - """Whether the component is a core component or not. - - Returns: - bool: Whether this is a core component. - """ - pass - - def __call__(self, doc: MutableDocument) -> MutableDocument: - pass - - @classmethod - def create_new_component( - cls, cnf: ComponentConfig, tokenizer: BaseTokenizer, - cdb: CDB, vocab: Vocab, model_load_path: Optional[str]) -> Self: - """Create a new component or load one off disk if load path presented. - - This may raise an exception if the wrong type of config is provided. - - Args: - cnf (ComponentConfig): The config relevant to this components. - tokenizer (BaseTokenizer): The base tokenizer. - cdb (CDB): The CDB. - vocab (Vocab): The Vocab. - model_load_path (Optional[str]): Model load path (if present). - - Returns: - Self: The new components. - """ - pass +from medcat.components.base import BaseComponent, CoreComponentType @runtime_checkable diff --git a/medcat-v2/medcat/pipeline/pipeline.py b/medcat-v2/medcat/pipeline/pipeline.py index 5e24bc421..6e334b8f3 100644 --- a/medcat-v2/medcat/pipeline/pipeline.py +++ b/medcat-v2/medcat/pipeline/pipeline.py @@ -341,8 +341,33 @@ def get_doc(self, text: str) -> MutableDocument: Returns: MutableDocument: The resulting document. """ + return self.pipe_until(text, None) + + def pipe_until( + self, + text: str, + comp_type: Optional[CoreComponentType] + ) -> MutableDocument: + """Run the pipe until the specific component (excluded). + + If `comp_type == None` then the entire pipe is run. + Otherwise the pipe is stopped before the specificied component is run. + + Args: + text (str): The text to run over. + comp_type (Optional[CoreComponentType]): The last component to run or None. + + Returns: + MutableDocument: The processed document. + """ doc = self._tokenizer(text) for comp in self._components: + if comp_type and comp.get_type() == comp_type: + logger.info( + "Finishing pipe before %s (%s) as requested", + comp.get_type().name, comp.full_name + ) + return doc logger.info("Running component %s for %d of text (%s)", comp.full_name, len(text), id(text)) doc = comp(doc) @@ -350,6 +375,7 @@ def get_doc(self, text: str) -> MutableDocument: doc = addon(doc) return doc + def entity_from_tokens(self, tokens: list[MutableToken]) -> MutableEntity: """Get the entity from the list of tokens. diff --git a/medcat-v2/tests/components/test_contracting.py b/medcat-v2/tests/components/test_contracting.py new file mode 100644 index 000000000..225779a30 --- /dev/null +++ b/medcat-v2/tests/components/test_contracting.py @@ -0,0 +1,21 @@ +from unittest import TestCase + +from medcat.cat import CAT +from medcat.components import contracting_testing +from medcat.components.base import CoreComponentType +from tests import UNPACKED_EXAMPLE_MODEL_PACK_PATH + + +class TestContractingForModel(TestCase): + + @classmethod + def setUpClass(cls) -> None: + cls._model = CAT.load_model_pack(UNPACKED_EXAMPLE_MODEL_PACK_PATH) + + def test_all_contracts_hold(self): + contracting_testing.assert_component_contracts(self._model) + + def test_individual_contracts_hold(self): + for ct in [CoreComponentType.ner, CoreComponentType.linking]: + comp = self._model.pipe.get_component(ct) + contracting_testing.assert_single_component_holds(self._model, comp) diff --git a/medcat-v2/tests/components/test_contracting_testing.py b/medcat-v2/tests/components/test_contracting_testing.py new file mode 100644 index 000000000..df96ae019 --- /dev/null +++ b/medcat-v2/tests/components/test_contracting_testing.py @@ -0,0 +1,39 @@ +from unittest import TestCase +from contextlib import contextmanager + +from medcat.cat import CAT +from medcat.components import contracting_testing +from medcat.utils.cdb_state import captured_state_cdb +from tests import UNPACKED_EXAMPLE_MODEL_PACK_PATH + + +class ContractingTestingTests(TestCase): + + @classmethod + def setUpClass(cls) -> None: + cls._model = CAT.load_model_pack(UNPACKED_EXAMPLE_MODEL_PACK_PATH) + + @contextmanager + def empty_cdb(self): + with captured_state_cdb(self._model.cdb): + self._model.cdb.name2info.clear() + self._model.cdb.cui2info.clear() + yield + + def test_contracting_normally_passes(self): + contracting_testing.assert_component_contracts(self._model) + + def test_contracting_fails_with_empty_cdb(self): + with self.empty_cdb(): + with self.assertRaises(contracting_testing.ContractViolationError): + contracting_testing.assert_component_contracts(self._model) + + def test_contracting_fails_with_no_entity(self): + with self.assertRaises(contracting_testing.ContractViolationError): + contracting_testing.assert_component_contracts( + self._model, "Text with no entities") + + def test_contracting_fails_with_no_tokens(self): + with self.assertRaises(contracting_testing.ContractViolationError): + contracting_testing.assert_component_contracts( + self._model, "")