Files
wehub-resource-sync dc6079821b
CI Github Actions / Run Tests (push) Waiting to run
Automatic PR Merger / mergepal (push) Waiting to run
Semgrep / Semgrep Workflow Security Scan (push) Waiting to run
Docs Tests / Check for file changes (push) Has been cancelled
Docs Tests / Test Documentation (push) Has been cancelled
Docs Tests / Documentation Linting Checks (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.10, test-performance) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.10, test-policies) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.8, test-cli) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.8, test-core-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.8, test-full-model-training) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.8, test-nlu-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.8, test-nlu-predictors) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.8, test-other-unit-tests) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.8, test-performance) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.8, test-policies) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.9, test-cli) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.9, test-core-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.9, test-full-model-training) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.9, test-nlu-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.9, test-nlu-predictors) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.9, test-other-unit-tests) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.9, test-performance) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.9, test-policies) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.10, test-cli) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.10, test-core-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.10, test-full-model-training) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.10, test-nlu-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.10, test-nlu-predictors) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.10, test-other-unit-tests) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.10, test-performance) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.10, test-policies) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.8, test-cli) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.8, test-core-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.8, test-full-model-training) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.8, test-nlu-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.8, test-nlu-predictors) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.8, test-other-unit-tests) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.8, test-performance) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.8, test-policies) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.9, test-cli) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.9, test-core-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.9, test-full-model-training) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.9, test-nlu-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.9, test-nlu-predictors) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.9, test-other-unit-tests) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.9, test-performance) (push) Has been cancelled
Continuous Integration / Run Tests (windows-2022, 3.9, test-policies) (push) Has been cancelled
Continuous Integration / Run Flaky Tests (ubuntu-24.04, 3.10) (push) Has been cancelled
Continuous Integration / Run Flaky Tests (ubuntu-24.04, 3.8) (push) Has been cancelled
Continuous Integration / Run Flaky Tests (ubuntu-24.04, 3.9) (push) Has been cancelled
Continuous Integration / Run Flaky Tests (windows-2022, 3.10) (push) Has been cancelled
Continuous Integration / Run Flaky Tests (windows-2022, 3.8) (push) Has been cancelled
Continuous Integration / Run Flaky Tests (windows-2022, 3.9) (push) Has been cancelled
Continuous Integration / Check for file changes (push) Has been cancelled
Continuous Integration / Wait for docs tests (push) Has been cancelled
Continuous Integration / Code Quality (push) Has been cancelled
Continuous Integration / Check for changelog (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.10, test-cli) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.10, test-core-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.10, test-full-model-training) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.10, test-nlu-featurizers) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.10, test-nlu-predictors) (push) Has been cancelled
Continuous Integration / Run Tests (ubuntu-24.04, 3.10, test-other-unit-tests) (push) Has been cancelled
Continuous Integration / Upload coverage reports to codeclimate (push) Has been cancelled
Continuous Integration / Run Non-Sequential Integration Tests (push) Has been cancelled
Continuous Integration / Run Broker Integration Tests (push) Has been cancelled
Continuous Integration / Run Sequential Integration Tests (push) Has been cancelled
Continuous Integration / Build Docker base images and setup environment (push) Has been cancelled
Continuous Integration / Build Docker (default) (push) Has been cancelled
Continuous Integration / Build Docker (full) (push) Has been cancelled
Continuous Integration / Build Docker (mitie-en) (push) Has been cancelled
Continuous Integration / Build Docker (spacy-de) (push) Has been cancelled
Continuous Integration / Build Docker (spacy-en) (push) Has been cancelled
Continuous Integration / Build Docker (spacy-it) (push) Has been cancelled
Continuous Integration / Deploy to PyPI (push) Has been cancelled
Continuous Integration / Notify Slack & Publish Release Notes (push) Has been cancelled
Publish Documentation / Evaluate release tag (push) Has been cancelled
Publish Documentation / Prebuild Docs (push) Has been cancelled
Publish Documentation / Preview Docs (push) Has been cancelled
Publish Documentation / Check for file changes (push) Has been cancelled
Publish Documentation / Publish Docs (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:24:47 +08:00

1119 lines
30 KiB
Python

import itertools
import shutil
import re
import textwrap
from pathlib import Path
from typing import Text, Any, Tuple, Callable
from unittest.mock import patch, call
import pytest
import rasa.shared.utils.io
from rasa.core import migrate
from rasa.shared.core.domain import Domain
from rasa.shared.exceptions import RasaException
from rasa.shared.constants import LATEST_TRAINING_DATA_FORMAT_VERSION
def prepare_domain_path(directory: Path, domain_content: Text, file_name: Text) -> Path:
original_content = textwrap.dedent(domain_content)
domain_file = directory / file_name
rasa.shared.utils.io.write_text_file(original_content, domain_file)
return domain_file
@pytest.fixture()
def domain_out_file(tmp_path: Path) -> Path:
return tmp_path / "custom_new_domain.yml"
def test_migrate_domain_format_with_required_slots(
tmp_path: Path, domain_out_file: Path
):
existing_domain_file = prepare_domain_path(
tmp_path,
"""
version: "2.0"
intents:
- greet
- affirm
- inform
entities:
- city
- name
slots:
location:
type: text
influence_conversation: false
name:
type: text
influence_conversation: false
auto_fill: false
email:
type: text
influence_conversation: false
forms:
booking_form:
ignored_intents:
- greet
required_slots:
location:
- type: from_entity
entity: city
email:
- type: from_text
intent: inform
name:
- type: from_entity
entity: surname
""",
"domain.yml",
)
migrate.migrate_domain_format(existing_domain_file, domain_out_file)
domain = Domain.from_path(domain_out_file)
assert domain
old_domain_path = tmp_path / "original_domain.yml"
assert old_domain_path
migrated_domain = rasa.shared.utils.io.read_yaml_file(domain_out_file)
migrated_training_data_version = migrated_domain.get("version")
assert migrated_training_data_version == LATEST_TRAINING_DATA_FORMAT_VERSION
migrated_slots = migrated_domain.get("slots")
expected_slots = {
"location": {
"type": "text",
"influence_conversation": False,
"mappings": [
{
"type": "from_entity",
"entity": "city",
"conditions": [{"active_loop": "booking_form"}],
}
],
},
"name": {
"type": "text",
"influence_conversation": False,
"mappings": [
{
"type": "from_entity",
"entity": "surname",
"conditions": [{"active_loop": "booking_form"}],
}
],
},
"email": {
"type": "text",
"influence_conversation": False,
"mappings": [
{
"type": "from_text",
"intent": "inform",
"conditions": [
{"active_loop": "booking_form", "requested_slot": "email"}
],
}
],
},
}
assert migrated_slots == expected_slots
migrated_forms = migrated_domain.get("forms")
expected_forms = {
"booking_form": {
"ignored_intents": ["greet"],
"required_slots": ["location", "email", "name"],
}
}
assert migrated_forms == expected_forms
def test_migrate_domain_form_without_required_slots(
tmp_path: Path, domain_out_file: Path
):
existing_domain_file = prepare_domain_path(
tmp_path,
"""
version: "2.0"
intents:
- greet
- affirm
- inform
entities:
- city
- name
- surname
slots:
location:
type: text
influence_conversation: false
name:
type: text
influence_conversation: false
auto_fill: false
email:
type: text
influence_conversation: false
forms:
booking_form:
ignored_intents:
- greet
location:
- type: from_entity
entity: city
email:
- type: from_text
intent: inform
name:
- type: from_entity
entity: surname
""",
"domain.yml",
)
migrate.migrate_domain_format(existing_domain_file, domain_out_file)
domain = Domain.from_path(domain_out_file)
assert domain
old_domain_path = tmp_path / "original_domain.yml"
assert old_domain_path
migrated_domain = rasa.shared.utils.io.read_yaml_file(domain_out_file)
migrated_slots = migrated_domain.get("slots")
expected_slots = {
"location": {
"type": "text",
"influence_conversation": False,
"mappings": [
{
"type": "from_entity",
"entity": "city",
"conditions": [{"active_loop": "booking_form"}],
}
],
},
"name": {
"type": "text",
"influence_conversation": False,
"mappings": [
{
"type": "from_entity",
"entity": "surname",
"conditions": [{"active_loop": "booking_form"}],
}
],
},
"email": {
"type": "text",
"influence_conversation": False,
"mappings": [
{
"type": "from_text",
"intent": "inform",
"conditions": [
{"active_loop": "booking_form", "requested_slot": "email"}
],
}
],
},
}
assert migrated_slots == expected_slots
migrated_forms = migrated_domain.get("forms")
expected_forms = {
"booking_form": {
"ignored_intents": ["greet"],
"required_slots": ["location", "email", "name"],
}
}
assert migrated_forms == expected_forms
@pytest.mark.parametrize(
"slot_type,value",
[
("bool", True),
("float", 1),
("text", "out"),
("categorical", "test"),
("list", ["out"]),
("any", "etc"),
],
)
def test_migrate_domain_with_diff_slot_types(
slot_type: Text, value: Any, tmp_path: Path, domain_out_file: Path
):
existing_domain_file = prepare_domain_path(
tmp_path,
f"""
version: "2.0"
entities:
- outdoor
slots:
outdoor_seating:
type: {slot_type}
influence_conversation: false
forms:
reservation_form:
required_slots:
outdoor_seating:
- type: from_intent
value: {value}
intent: confirm
""",
"domain.yml",
)
migrate.migrate_domain_format(existing_domain_file, domain_out_file)
domain = Domain.from_path(domain_out_file)
assert domain
migrated_domain = rasa.shared.utils.io.read_yaml_file(domain_out_file)
migrated_slots = migrated_domain.get("slots")
expected_slots = {
"outdoor_seating": {
"type": slot_type,
"influence_conversation": False,
"mappings": [
{
"type": "from_intent",
"value": value,
"intent": "confirm",
"conditions": [
{
"active_loop": "reservation_form",
"requested_slot": "outdoor_seating",
}
],
}
],
}
}
assert migrated_slots == expected_slots
def test_migrate_domain_format_from_dir(tmp_path: Path):
domain_dir = tmp_path / "domain"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
entities:
- outdoor
slots:
outdoor_seating:
type: bool
influence_conversation: false
""",
"slots.yml",
)
prepare_domain_path(
domain_dir,
"""
version: "2.0"
forms:
reservation_form:
required_slots:
outdoor_seating:
- type: from_intent
value: true
intent: confirm
""",
"forms.yml",
)
domain_out_dir = tmp_path / "new_domain"
domain_out_dir.mkdir()
migrate.migrate_domain_format(domain_dir, domain_out_dir)
domain = Domain.from_directory(str(domain_out_dir))
assert domain
old_domain_path = tmp_path / "original_domain"
assert old_domain_path.exists()
for file in old_domain_path.iterdir():
assert file.name in ["slots.yml", "forms.yml"]
for file in domain_out_dir.iterdir():
assert file.name in ["slots.yml", "forms.yml"]
migrated_file = rasa.shared.utils.io.read_yaml_file(file)
migrated_training_data_version = migrated_file.get("version")
assert migrated_training_data_version == LATEST_TRAINING_DATA_FORMAT_VERSION
def test_migrate_domain_all_keys(tmp_path: Path, domain_out_file: Path):
existing_domain_file = prepare_domain_path(
tmp_path,
"""
version: "2.0"
intents:
- greet
entities:
- city
slots:
city:
type: text
influence_conversation: false
responses:
utter_greet:
- text: "Hi there!"
actions:
- action_check_time
forms:
booking_form:
required_slots:
city:
- type: from_entity
entity: city
""",
"domain.yml",
)
migrate.migrate_domain_format(existing_domain_file, domain_out_file)
domain = Domain.from_path(domain_out_file)
assert domain
migrated_domain = rasa.shared.utils.io.read_yaml_file(domain_out_file)
migrated_intents = migrated_domain.get("intents")
assert "greet" in migrated_intents
migrated_entities = migrated_domain.get("entities")
assert "city" in migrated_entities
migrated_responses = migrated_domain.get("responses")
assert "utter_greet" in migrated_responses
migrated_actions = migrated_domain.get("actions")
assert "action_check_time" in migrated_actions
migrated_training_data_version = migrated_domain.get("version")
assert migrated_training_data_version == LATEST_TRAINING_DATA_FORMAT_VERSION
def test_migrate_domain_format_with_custom_slot(tmp_path: Path, domain_out_file: Path):
existing_domain_file = prepare_domain_path(
tmp_path,
"""
version: "2.0"
intents:
- greet
- affirm
- inform
entities:
- city
- name
slots:
location:
type: text
influence_conversation: false
name:
type: text
influence_conversation: false
auto_fill: false
email:
type: text
influence_conversation: false
forms:
booking_form:
ignored_intents:
- greet
required_slots:
location:
- type: from_entity
entity: city
email:
- type: from_text
intent: inform
""",
"domain.yml",
)
with pytest.warns(UserWarning, match="A custom mapping was added to slot 'name'."):
migrate.migrate_domain_format(existing_domain_file, domain_out_file)
domain = Domain.from_path(domain_out_file)
assert domain
migrated_domain = rasa.shared.utils.io.read_yaml_file(domain_out_file)
migrated_slots = migrated_domain.get("slots")
custom_slot = migrated_slots.get("name")
assert custom_slot == {
"type": "text",
"influence_conversation": False,
"mappings": [{"type": "custom"}],
}
def test_migrate_domain_with_no_requested_slot_for_from_entity_mappings(
tmp_path: Path, domain_out_file: Path
):
existing_domain_file = prepare_domain_path(
tmp_path,
"""
version: "2.0"
intents:
- greet
- affirm
- inform
entities:
- city
slots:
location:
type: text
influence_conversation: false
email:
type: text
influence_conversation: false
forms:
some_form:
location:
- entity: city
type: from_entity
- intent: something
type: from_text
""",
"domain.yml",
)
migrate.migrate_domain_format(existing_domain_file, domain_out_file)
domain = Domain.from_path(domain_out_file)
assert domain
migrated_domain = rasa.shared.utils.io.read_yaml_file(domain_out_file)
migrated_slots = migrated_domain.get("slots")
location_slot = migrated_slots.get("location")
mappings = location_slot.get("mappings")
assert mappings[0] == {
"entity": "city",
"type": "from_entity",
"conditions": [{"active_loop": "some_form"}],
}
assert mappings[1] == {
"intent": "something",
"type": "from_text",
"conditions": [{"active_loop": "some_form", "requested_slot": "location"}],
}
def test_migrate_domain_format_duplicated_slots_in_forms(
tmp_path: Path, domain_out_file: Path
):
existing_domain_file = prepare_domain_path(
tmp_path,
"""
version: "2.0"
intents:
- greet
- affirm
- inform
entities:
- city
slots:
name:
type: text
influence_conversation: false
location:
type: text
influence_conversation: false
forms:
form_one:
required_slots:
name:
- type: from_text
intent: inform
location:
- type: from_text
intent: greet
form_two:
required_slots:
name:
- type: from_text
intent: inform
- type: from_intent
intent: deny
value: demo
location:
- type: from_entity
entity: city
""",
"domain.yml",
)
migrate.migrate_domain_format(existing_domain_file, domain_out_file)
domain = Domain.from_path(domain_out_file)
assert domain
migrated_domain = rasa.shared.utils.io.read_yaml_file(domain_out_file)
migrated_slots = migrated_domain.get("slots")
slot_with_duplicate_mappings = migrated_slots.get("name")
assert slot_with_duplicate_mappings == {
"type": "text",
"influence_conversation": False,
"mappings": [
{
"type": "from_text",
"intent": "inform",
"conditions": [
{"active_loop": "form_one", "requested_slot": "name"},
{"active_loop": "form_two", "requested_slot": "name"},
],
},
{
"type": "from_intent",
"intent": "deny",
"value": "demo",
"conditions": [{"active_loop": "form_two", "requested_slot": "name"}],
},
],
}
slot_with_different_mapping_conditions = migrated_slots.get("location")
assert slot_with_different_mapping_conditions == {
"type": "text",
"influence_conversation": False,
"mappings": [
{
"type": "from_text",
"intent": "greet",
"conditions": [
{"active_loop": "form_one", "requested_slot": "location"}
],
},
{
"type": "from_entity",
"entity": "city",
"conditions": [{"active_loop": "form_two"}],
},
],
}
def test_migrate_domain_dir_with_out_path_None(tmp_path: Path):
domain_dir = tmp_path / "domain"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
entities:
- outdoor
slots:
outdoor_seating:
type: bool
influence_conversation: false
""",
"slots.yml",
)
prepare_domain_path(
domain_dir,
"""
version: "2.0"
forms:
reservation_form:
required_slots:
outdoor_seating:
- type: from_intent
value: true
intent: confirm
""",
"forms.yml",
)
migrate.migrate_domain_format(domain_dir, None)
domain_out = tmp_path / "new_domain"
assert domain_out.exists()
for file in domain_out.iterdir():
assert file.name in ["slots.yml", "forms.yml"]
domain = Domain.from_directory(str(domain_out))
assert domain
def test_migrate_domain_multiple_files_with_duplicate_slots(tmp_path: Path):
domain_dir = tmp_path / "domain"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
entities:
- outdoor
slots:
outdoor_seating:
type: bool
influence_conversation: false
""",
"slots_one.yml",
)
prepare_domain_path(
domain_dir,
"""
version: "2.0"
entities:
- outdoor
slots:
cuisine:
type: text
influence_conversation: false
""",
"slots_two.yml",
)
with pytest.raises(
RasaException,
match="Domain files with multiple 'slots' sections were provided.",
):
migrate.migrate_domain_format(domain_dir, None)
def test_migrate_domain_with_multiple_files_with_duplicate_forms(tmp_path: Path):
domain_dir = tmp_path / "domain"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
forms:
reservation_form:
required_slots:
outdoor_seating:
- type: from_intent
value: true
intent: confirm
""",
"forms_one.yml",
)
prepare_domain_path(
domain_dir,
"""
version: "2.0"
forms:
reservation_form:
required_slots:
cuisine:
- type: from_entity
entity: cuisine
""",
"forms_two.yml",
)
with pytest.raises(
RasaException,
match="Domain files with multiple 'forms' sections were provided.",
):
migrate.migrate_domain_format(domain_dir, None)
def test_migrate_domain_from_dir_with_other_sections(tmp_path: Path):
domain_dir = tmp_path / "domain"
domain_dir.mkdir()
domain_file_one = "domain_one.yml"
domain_file_two = "domain_two.yml"
prepare_domain_path(
domain_dir,
"""
version: "2.0"
entities:
- outdoor
slots:
outdoor_seating:
type: bool
influence_conversation: false
""",
domain_file_one,
)
prepare_domain_path(
domain_dir,
"""
version: "2.0"
intents:
- greet
forms:
reservation_form:
required_slots:
outdoor_seating:
- type: from_intent
value: true
intent: confirm
""",
domain_file_two,
)
new_domain_dir = tmp_path / "migrated_domain"
migrate.migrate_domain_format(domain_dir, new_domain_dir)
domain = Domain.from_directory(new_domain_dir)
assert domain
for file in new_domain_dir.iterdir():
migrated = rasa.shared.utils.io.read_yaml_file(file)
migrated_training_data_version = migrated.get("version")
assert migrated_training_data_version == LATEST_TRAINING_DATA_FORMAT_VERSION
if file.name == domain_file_one:
assert migrated.get("entities") == ["outdoor"]
elif file.name == domain_file_two:
assert migrated.get("intents") == ["greet"]
def test_migrate_domain_raises_exception_for_non_domain_file(tmp_path: Path):
domain_file = prepare_domain_path(
tmp_path,
"""
version: "2.0"
nlu:
- intent: greet
examples: |
- hey
- hello
- hi
- hello there
- good morning
- good evening
- moin
- hey there
""",
"domain.yml",
)
new_domain_file = tmp_path / "new_domain.yml"
with pytest.raises(
RasaException,
match=f"The file '{domain_file.as_posix()}' could not "
f"be validated as a domain file.",
):
migrate.migrate_domain_format(domain_file, new_domain_file)
def test_migrate_domain_raises_for_non_domain_files(tmp_path: Path):
domain_dir = tmp_path / "domain"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
nlu:
- intent: greet
examples: |
- hey
- hello
- hi
- hello there
- good morning
- good evening
- moin
- hey there
""",
"domain.yml",
)
with pytest.raises(
RasaException,
match=f"The domain directory '{domain_dir.as_posix()}' does not contain any "
f"domain files. Please make sure to include these for a successful "
f"migration.",
):
migrate.migrate_domain_format(domain_dir, None)
def test_migrate_domain_raises_for_missing_slots_and_forms(tmp_path: Path):
domain_dir = tmp_path / "domain"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
entities:
- bla
""",
"domain.yml",
)
with pytest.raises(
RasaException,
match=f"The files you have provided in '{re.escape(str(domain_dir))}' "
f"are missing slots or forms. "
f"Please make sure to include these for a successful migration.",
):
migrate.migrate_domain_format(domain_dir, None)
def test_migrate_domain_raises_when_migrated_files_are_found(tmp_path: Path):
domain_dir = tmp_path / "domain"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
f"""
version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}"
intents: []
""",
"domain.yml",
)
with pytest.raises(
RasaException, match="Some of the given files (.*) have already been migrated.*"
):
migrate.migrate_domain_format(domain_dir, None)
def test_migrate_folder_only_migrates_domain_files(tmp_path: Path):
domain_dir = tmp_path / "domain"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
entities:
- outdoor
slots:
outdoor_seating:
type: bool
influence_conversation: false
""",
"slots.yml",
)
prepare_domain_path(
domain_dir,
"""
version: "2.0"
forms:
reservation_form:
required_slots:
outdoor_seating:
- type: from_intent
value: true
intent: confirm
""",
"forms.yml",
)
prepare_domain_path(
domain_dir,
"""
not a domain file.
""",
"not-a-domain-file.yml",
)
out_dir = tmp_path / "out_dir"
migrate.migrate_domain_format(domain_dir, out_dir)
assert set(f.name for f in out_dir.iterdir()) == {"forms.yml", "slots.yml"}
# i.e. the not-a-domain-file is not migrated
def example_migrate_folder_fails_because_multiple_slots_sections(
path: Path,
) -> Tuple[Path, Text]:
domain_dir = path / "domain"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
entities:
- outdoor
slots:
outdoor_seating:
type: bool
influence_conversation: false
""",
"slots_one.yml",
)
prepare_domain_path(
domain_dir,
"""
version: "2.0"
entities:
- outdoor
slots:
cuisine:
type: text
influence_conversation: false
""",
"slots_two.yml",
)
prepare_domain_path(
domain_dir,
"""
version: "2.0"
responses:
utter_greet:
- text: "Hi there!"
""",
"responses.yml",
)
return domain_dir, "Domain files with multiple 'slots' sections were " "provided."
@pytest.mark.parametrize(
"test_case, default_out_dir, out_dir_exists",
[
(test_case, default_out_dir, out_dir_exists)
for test_case in [example_migrate_folder_fails_because_multiple_slots_sections]
for (default_out_dir, out_dir_exists) in [
(True, True),
(True, False),
(False, False),
]
],
)
def test_migrate_domain_cleanups_after_raising(
tmp_path: Path,
test_case: Callable[[Path], Tuple[Text, Text]],
default_out_dir: bool,
out_dir_exists: bool,
):
# input
out_path = None if default_out_dir else (tmp_path / "custom_out_path")
domain_path, error_msg_match = test_case(tmp_path)
migrating_file_only = domain_path.is_file()
domain_files = list(domain_path.iterdir())
# paths to be used by migration tool
domain_parent_dir = domain_path.parent
expected_out_path = (
out_path
if out_path is not None
else (domain_parent_dir / migrate.DEFAULT_NEW_DOMAIN)
)
expected_backup_path = domain_parent_dir / migrate.ORIGINAL_DOMAIN
if migrating_file_only:
expected_backup_path = f"{expected_backup_path}{migrate.YML_SUFFIX}"
expected_out_path = f"{expected_out_path}{migrate.YML_SUFFIX}"
# create the folder if needed
if not migrating_file_only and out_dir_exists:
expected_out_path.mkdir(parents=True)
# migrate!
with pytest.raises(RasaException, match=error_msg_match):
migrate.migrate_domain_format(domain_path, out_path)
assert Path.exists(domain_path)
assert all(Path.exists(file) for file in domain_files)
assert not Path.exists(expected_backup_path)
if not migrating_file_only:
# if and only if the folder didn't exist before, it should not exist afterwards
assert Path.exists(expected_out_path) == out_dir_exists
# ... and to assert we really did remove something:
expected_to_be_removed = [call(expected_backup_path)]
if not out_dir_exists: # only removed if it didn't exist
expected_to_be_removed.append(call(expected_out_path))
patching = [shutil, "rmtree"] if not migrating_file_only else [Path, "unlink"]
with patch.object(*patching) as removal:
with pytest.raises(RasaException, match=error_msg_match):
migrate.migrate_domain_format(domain_path, out_path)
assert removal.call_count == 1 + (not out_dir_exists)
removal.assert_has_calls(expected_to_be_removed)
@pytest.mark.parametrize("migrate_file_only", [True, False])
def test_migrate_domain_raises_when_backup_location_exists(
tmp_path: Path, migrate_file_only: bool
):
domain_dir = tmp_path / "domain"
domain_file_name = "domain.yml"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
intents: []
""",
domain_file_name,
)
if not migrate_file_only:
domain_path = domain_dir
backup_location = tmp_path / migrate.ORIGINAL_DOMAIN
backup_location.mkdir()
else:
domain_path = domain_dir / domain_file_name
backup_location = domain_dir / (migrate.ORIGINAL_DOMAIN + migrate.YML_SUFFIX)
with open(backup_location, "w"):
pass
with pytest.raises(
RasaException, match="The domain could not be migrated since .* exists.*"
):
migrate.migrate_domain_format(domain_path, None)
@pytest.mark.parametrize(
"migrate_file_only, default_location", itertools.combinations([True, False], 2)
)
def test_migrate_domain_raises_when_output_location_is_used(
tmp_path: Path, migrate_file_only: bool, default_location: bool
):
domain_dir = tmp_path / "domain"
domain_file_name = "domain.yml"
domain_dir.mkdir()
prepare_domain_path(
domain_dir,
"""
version: "2.0"
intents: []
""",
domain_file_name,
)
if not migrate_file_only:
domain_path = domain_dir
if default_location:
out_path = None
non_empty_existing_dir = tmp_path / migrate.DEFAULT_NEW_DOMAIN
else:
out_path = tmp_path / "my_custom_new_domain_name"
non_empty_existing_dir = out_path
non_empty_existing_dir.mkdir()
# in contrast to the backup location, the output directory may exist
# but must be empty
with open(non_empty_existing_dir / "bla.txt", "w"):
pass
else:
domain_path = domain_dir / domain_file_name
if default_location:
out_path = None
existing_file = tmp_path / (migrate.DEFAULT_NEW_DOMAIN + migrate.YML_SUFFIX)
else:
out_path = tmp_path / "my_custom_file_name.yml"
existing_file = out_path
with open(existing_file, "w"):
pass
with pytest.raises(
RasaException,
match="The domain could not be migrated to .* because .* already exists.*",
):
migrate.migrate_domain_format(domain_path, out_path)