Repository: SFDO-Tooling/CumulusCI
Pull request #3473 (2022-12-17 02:19:45)

Refresh all objects in org schema cache
Previously the org schema was versioned as a unified whole. Not one sobject at a time. This worked until I added the feature of allowing one to download a subset of the schema. Now some objects might be refreshed on Monday, some on Tuesday, a different set on Wednesday etc. To allow this, all objects are independently versioned (with a last_modified_date) now. If it is needed by the application it is refreshed and if it isn't, it isn't.

This bug was hidden because of the ways we were using `get_org_schema` before we added `capture_sample_data`. Now we really need the schema to be complete and up-to-date.

Tasks that use `get_org_schema` include `load_data` (for the Recently Used checking) and `generate_mapping`.




Patch:
diff --git a/cumulusci/salesforce_api/org_schema.py b/cumulusci/salesforce_api/org_schema.py
index 7097f3312..4b2bc03b3 100644
--- a/cumulusci/salesforce_api/org_schema.py
+++ b/cumulusci/salesforce_api/org_schema.py
@@ -3,12 +3,11 @@ import re
 import typing as T
 from collections import defaultdict
 from contextlib import ExitStack, contextmanager
-from email.utils import parsedate
 from enum import Enum
 from logging import getLogger
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import Dict, Iterable, List, NamedTuple, Optional, Tuple
+from typing import Dict, Iterable, List, NamedTuple, Optional
 
 from sqlalchemy import MetaData, create_engine, not_
 from sqlalchemy.engine import Engine
@@ -112,7 +111,6 @@ class Filters(Enum):
 class Schema:
     """Represents an org's schema, cached from describe() calls"""
 
-    _last_modified_date = None
     included_objects = None
     includes_counts = False
 
@@ -164,21 +162,6 @@ class Schema:
     def close(self):
         self.session.close()
 
-    @property
-    def last_modified_date(self):
-        """Date of the most recent schema update"""
-        if not self._last_modified_date:
-            try:
-                self._last_modified_date = (
-                    self.session.query(FileMetadata)
-                    .filter(FileMetadata.name == "Last-Modified")
-                    .one()
-                    .value
-                )
-            except exc.NoResultFound:
-                pass
-        return self._last_modified_date
-
     def __repr__(self):
         return f"<Schema {self.path} : {self.engine}>"
 
@@ -193,7 +176,6 @@ class Schema:
         self,
         sf,
         included_objects,
-        last_modified_date,
         filters: T.Sequence[Filters] = (),
         patterns_to_ignore: T.Sequence[str] = (),
         logger=None,
@@ -223,18 +205,26 @@ class Schema:
                 or ignore_based_on_properties(obj, filters)
             )
         ]
-        changes = list(deep_describe(sf, last_modified_date, sobj_names, logger))
 
-        self._populate_cache_from_describe(changes, last_modified_date)
+        def find_last_modified_date(sobj_name: str) -> Optional[str]:
+            if obj := self.get(sobj_name):
+                return obj.last_modified_date
+            else:
+                return y2k
+
+        objs_to_refresh = {
+            sobj_name: find_last_modified_date(sobj_name) for sobj_name in sobj_names
+        }
+        changes = list(deep_describe(sf, objs_to_refresh, logger))
+
+        self._populate_cache_from_describe(changes)
         if include_counts:
             results = populate_counts(sf, self, sobj_names, logger)
         else:
             results = {name: None for name in sobj_names}
         return results
 
-    def _populate_cache_from_describe(
-        self, describe_objs: List[Tuple[dict, str]], last_modified_date
-    ) -> T.List[str]:
+    def _populate_cache_from_describe(self, describe_objs: List["DescribeUpdate"]):
         """Populate a schema cache from a list of describe objects."""
         engine = self.engine
         metadata = Base.metadata
@@ -243,32 +233,47 @@ class Schema:
 
         with BufferedSession(engine, metadata) as sess:
 
-            max_last_modified = (parsedate(last_modified_date), last_modified_date)
             for (sobj_data, last_modified) in describe_objs:
                 sobj_data = sobj_data.copy()
                 fields = sobj_data.pop("fields")
+                sobj_data["last_modified_date"] = last_modified
                 create_row(sess, SObject, sobj_data)
                 for field in fields:
                     field["sobject"] = sobj_data["name"]
                     create_row(sess, Field, field)
-                    sortable = parsedate(last_modified), last_modified
-                    if sortable > max_last_modified:
-                        max_last_modified = sortable
-
-            create_row(
-                sess,
-                FileMetadata,
-                {"name": "Last-Modified", "value": max_last_modified[1]},
-            )
-            create_row(sess, FileMetadata, {"name": "FormatVersion", "value": 1})
+
+            self.save_version(sess)
 
         engine.execute("vacuum")
 
+    FormatVersion = "FormatVersion"
+    CurrentFormatVersion = 2
+
+    @property
+    def version(self) -> int:
+        rows = self.session.query(FileMetadata.value)
+        rows = rows.filter(FileMetadata.name == self.FormatVersion)
+        first_row = rows.one()
+        assert first_row
+        return int(first_row[0])
+
+    def save_version(self, sess: "BufferedSession"):
+        create_row(
+            sess,
+            FileMetadata,
+            {"name": self.FormatVersion, "value": self.CurrentFormatVersion},
+        )
+        self.session.commit()
+
 
 def create_row(buffered_session: "BufferedSession", model, valuesdict: dict):
     buffered_session.write_single_row(model.__tablename__, valuesdict)
 
 
+class SilentMigration(Exception):
+    pass
+
+
 class BufferedSession:
     """Buffer writes to a SQL DB for faster performmance"""
 
@@ -334,7 +339,7 @@ def get_org_schema(
     *,
     include_counts: bool = False,
     filters: T.Sequence[Filters] = (),
-    patterns_to_ignore: T.Tuple[str] = (),
+    patterns_to_ignore: T.Tuple[str, ...] = (),
     included_objects: T.List[str] = (),
     force_recache=False,
     logger=None,
@@ -397,12 +402,17 @@ def get_org_schema(
 
                     cleanups_on_failure.append(schema.close)
                     closer.callback(schema.close)
+                    if schema.version != schema.CurrentFormatVersion:
+                        raise SilentMigration(
+                            "was created with older CumulusCI version"
+                        )
                     assert schema.sobjects.first().name
                     schema.from_cache = True
                 except Exception as e:
-                    logger.warning(
-                        f"Cannot read `{schema_path}`. Recreating it. Reason `{e}`."
-                    )
+                    if not isinstance(e, SilentMigration):
+                        logger.warning(
+                            f"Cannot read `{schema_path}`. Recreating it. Reason `{e}`."
+                        )
                     schema = None
                     for cleanup_action in reversed(cleanups_on_failure):
                         cleanup_action()
@@ -417,7 +427,6 @@ def get_org_schema(
             populated_objs = schema.populate_cache(
                 sf,
                 included_objects,
-                schema.last_modified_date or y2k,
                 filters,
                 patterns_to_ignore,
                 logger,
@@ -494,9 +503,10 @@ class DescribeUpdate(NamedTuple):
     last_modified_date: str = None
 
 
-def deep_describe(
-    sf, last_modified_date: Optional[str], objs: List[str], logger
-) -> Iterable[DescribeUpdate]:
+lm_date = Optional[str]
+
+
+def deep_describe(sf, objs: Dict[str, lm_date], logger) -> Iterable[DescribeUpdate]:
     """Fetch describe data for changed sobjects
 
     Fetch describe data for sobjects from the list 'objs'
@@ -504,7 +514,6 @@ def deep_describe(
     proto format) and yield each object as a DescribeResponse object."""
 
     logger = logger or getLogger(__name__)
-    last_modified_date = last_modified_date or y2k
     with CompositeParallelSalesforce(sf, max_workers=8) as cpsf:
         responses, errors = cpsf.do_composite_requests(
             (
@@ -512,9 +521,9 @@ def deep_describe(
                     "method": "GET",
                     "url": f"/services/data/v{sf.sf_version}/sobjects/{obj}/describe",
                     "referenceId": f"ref{obj}",
-                    "httpHeaders": {"If-Modified-Since": last_modified_date},
+                    "httpHeaders": {"If-Modified-Since": last_modified_date or y2k},
                 }
-                for obj in objs
+                for obj, last_modified_date in objs.items()
             )
         )
 
diff --git a/cumulusci/salesforce_api/org_schema_models.py b/cumulusci/salesforce_api/org_schema_models.py
index ba8cd68e6..3e191c76f 100644
--- a/cumulusci/salesforce_api/org_schema_models.py
+++ b/cumulusci/salesforce_api/org_schema_models.py
@@ -126,6 +126,7 @@ class SObject(OrgSchemaModelMixin, Base):
     supportedScopes = Column(SequenceType)
     actionOverrides = Column(SequenceType)
     count = Column(Integer)
+    last_modified_date = Column(String)
 
     @property
     def extractable(self):


Test patch:
diff --git a/cumulusci/conftest.py b/cumulusci/conftest.py
index f153f883b..cce385cde 100644
--- a/cumulusci/conftest.py
+++ b/cumulusci/conftest.py
@@ -16,10 +16,17 @@ from sqlalchemy.orm import sessionmaker
 from cumulusci.core.dependencies.utils import TaskContext
 from cumulusci.core.github import get_github_api
 from cumulusci.salesforce_api.org_schema_models import Base
+from cumulusci.tasks.bulkdata.tests.integration_test_utils import (
+    ensure_accounts,
+    ensure_records,
+)
 from cumulusci.tasks.salesforce.tests.util import create_task_fixture
 from cumulusci.tests.pytest_plugins.pytest_sf_vcr import salesforce_vcr, vcr_config
 from cumulusci.tests.util import DummyKeychain, DummyOrgConfig, mock_env
 
+ensure_accounts = ensure_accounts
+ensure_records = ensure_records
+
 
 @fixture(scope="session", autouse=True)
 def mock_sleep():
diff --git a/cumulusci/core/tests/test_datasets_e2e.py b/cumulusci/core/tests/test_datasets_e2e.py
index c0afab0ca..c0a7b7f8b 100644
--- a/cumulusci/core/tests/test_datasets_e2e.py
+++ b/cumulusci/core/tests/test_datasets_e2e.py
@@ -10,9 +10,13 @@ from cumulusci.tasks.bulkdata.extract_dataset_utils.tests.test_synthesize_extrac
     _fake_get_org_schema,
     describe_for,
 )
-from cumulusci.tasks.bulkdata.tests.integration_test_utils import ensure_accounts
+from cumulusci.tasks.bulkdata.tests.integration_test_utils import (
+    ensure_accounts,
+    ensure_records,
+)
 
 ensure_accounts = ensure_accounts
+ensure_records = ensure_records
 
 
 class Timer:
diff --git a/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_synthesize_extract_declarations.py b/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_synthesize_extract_declarations.py
index 43891f833..3da24b5c3 100644
--- a/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_synthesize_extract_declarations.py
+++ b/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_synthesize_extract_declarations.py
@@ -352,6 +352,43 @@ class TestSynthesizeExtractDeclarations:
             assert set(decls["Account"].fields) == set(["Name", "Description"])
             assert decls["Contact"].fields == ["LastName"], decls["Contact"].fields
 
+    def test_synthesize_fields(self, sf, org_config):
+        declarations = """
+            extract:
+                Opportunity:
+                    fields:
+                        - FIELDS(STANDARD)
+                Account:
+                    fields:
+                        - FIELDS(CUSTOM)
+                Contact:
+                    fields:
+                        - FIELDS(ALL)
+                Custom__c:
+                    fields:
+                        - FIELDS(CUSTOM)
+        """
+        declarations = ExtractRulesFile.parse_extract(StringIO(declarations))
+        object_counts = {"Account": 2, "Contact": 2, "Custom__c": 5}
+        object_describes = [describe_for(obj) for obj in object_counts.keys()]
+        with _fake_get_org_schema(
+            org_config,
+            object_describes,
+            object_counts,
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+            decls = {decl.sf_object: decl for decl in decls}
+            assert "Account" in decls
+            assert "Contact" in decls
+            assert "Entitlement" not in decls
+            assert "Opportunity" not in decls  # not populated
+            assert "Name" not in decls["Custom__c"].fields
+            assert "Id" not in decls["Custom__c"].fields
+            assert "Name" in decls["Account"].fields  # because required
+            assert "BillingCountry" not in decls["Account"].fields  # not required
+            assert "CustomField__c" in decls["Custom__c"].fields
+
     @pytest.mark.needs_org()
     @pytest.mark.slow()
     def test_find_standard_objects__integration_tests(self, sf, org_config):
@@ -372,7 +409,7 @@ class TestSynthesizeExtractDeclarations:
             decls = {decl.sf_object: decl for decl in decls}
             assert "WorkBadgeDefinition" in decls
             # HEY NOW!
-            assert "You\\'re a RockStar!" in decls["WorkBadgeDefinition"].where
+            assert "You\\'re a RockStar!" in str(decls["WorkBadgeDefinition"].where)
             if "Opportunity" in decls:
                 assert "IsPrivate" not in decls["Opportunity"].fields, decls.keys()
 
diff --git a/cumulusci/tasks/bulkdata/tests/integration_test_utils.py b/cumulusci/tasks/bulkdata/tests/integration_test_utils.py
index 8de89943a..416fc68c8 100644
--- a/cumulusci/tasks/bulkdata/tests/integration_test_utils.py
+++ b/cumulusci/tasks/bulkdata/tests/integration_test_utils.py
@@ -1,4 +1,5 @@
 from contextlib import contextmanager
+from typing import Dict, List
 
 import pytest
 
@@ -6,19 +7,43 @@ from cumulusci.tasks.bulkdata.delete import DeleteData
 
 
 @pytest.fixture()
-def ensure_accounts(create_task, run_code_without_recording, sf):
-    """Delete all accounts and create a certain number of new ones"""
+def ensure_records(create_task, run_code_without_recording, sf):
+    """Delete all records of an sobject and create a certain number of new ones"""
 
     @contextmanager
-    def _ensure_accounts(number_of_accounts):
-        def setup(number):
-            task = create_task(DeleteData, {"objects": "Entitlement, Account"})
+    def _ensure_records(sobjects: Dict[str, List[Dict[str, str]]]):
+        def delete():
+            task = create_task(DeleteData, {"objects": ",".join(sobjects.keys())})
             task()
-            for i in range(0, number):
-                sf.Account.create({"Name": f"Account {i}"})
 
-        run_code_without_recording(lambda: setup(number_of_accounts))
+        def setup():
+            for obj, records in sobjects.items():
+                proxy = getattr(sf, obj)
+                for record in records:
+                    proxy.create(record)
+
+        run_code_without_recording(delete)
+        run_code_without_recording(setup)
         yield
-        run_code_without_recording(lambda: setup(0))
+        run_code_without_recording(delete)
+
+    return _ensure_records
+
+
+@pytest.fixture()
+def ensure_accounts(ensure_records):
+    """Delete all accounts and create a certain number of new ones"""
+
+    @contextmanager
+    def _ensure_accounts(number_of_accounts):
+        with ensure_records(
+            {
+                "Entitlement": [],
+                "Account": [
+                    {"name": f"Account {i}"} for i in range(number_of_accounts)
+                ],
+            }
+        ):
+            yield
 
     return _ensure_accounts
diff --git a/cumulusci/tasks/bulkdata/tests/test_generatemapping.py b/cumulusci/tasks/bulkdata/tests/test_generatemapping.py
index 9bd36ea25..fa9a775a6 100644
--- a/cumulusci/tasks/bulkdata/tests/test_generatemapping.py
+++ b/cumulusci/tasks/bulkdata/tests/test_generatemapping.py
@@ -28,7 +28,7 @@ def _temp_schema_for_tests(describe_data: list):
         try:
             date = "Thu, 09 Feb 2021 21:35:07 GMT"
             describe_data_with_dates = [(dct, date) for dct in describe_data]
-            schema._populate_cache_from_describe(describe_data_with_dates, date)
+            schema._populate_cache_from_describe(describe_data_with_dates)
             yield schema
         finally:
             schema.close()
diff --git a/cumulusci/utils/tests/test_org_schema.py b/cumulusci/utils/tests/test_org_schema.py
index 42af5375f..34b664d60 100644
--- a/cumulusci/utils/tests/test_org_schema.py
+++ b/cumulusci/utils/tests/test_org_schema.py
@@ -17,11 +17,15 @@ from cumulusci.salesforce_api.org_schema import (
     zip_database,
 )
 from cumulusci.salesforce_api.org_schema_models import Base, SObject
-from cumulusci.tasks.bulkdata.tests.integration_test_utils import ensure_accounts
+from cumulusci.tasks.bulkdata.tests.integration_test_utils import (
+    ensure_accounts,
+    ensure_records,
+)
 from cumulusci.tests.util import FakeUnreliableRequestHandler
 from cumulusci.utils.http.multi_request import HTTPRequestError
 
 ensure_accounts = ensure_accounts  # fixes 4 lint errors at once. Don't hate the player, hate the game.
+ensure_records = ensure_records
 
 
 class FakeSF:
@@ -424,14 +428,63 @@ class TestOrgSchema:
             assert "Apostasy" in caplog.text
             assert "more counting errors suppressed" in caplog.text
 
+    def test_old_schema_version(self, sf, org_config, caplog):
+        with mock_return_uncached_responses(self.cassette_data):
+            with patch(
+                "cumulusci.salesforce_api.org_schema.Schema.CurrentFormatVersion", 7
+            ), get_org_schema(
+                FakeSF(), org_config, include_counts=True, filters=[Filters.populated]
+            ) as schema:
+                assert schema.version == 7
+
+            class FakeSilentMigration(Exception):
+                called = False
+
+                def __init__(self, *args, **kwargs):
+                    self.__class__.called = True
+
+            with patch(
+                "cumulusci.salesforce_api.org_schema.SilentMigration",
+                FakeSilentMigration,
+            ), patch(
+                "cumulusci.salesforce_api.org_schema.Schema.CurrentFormatVersion", 8
+            ), get_org_schema(
+                FakeSF(), org_config, include_counts=True, filters=[Filters.populated]
+            ) as schema:
+                assert schema.version == 8
+                assert FakeSilentMigration.called
+
     @pytest.mark.needs_org()
-    def test_schema_populated_real(self, sf, org_config, ensure_accounts):
-        with ensure_accounts(6):
+    def test_schema_populated_real(self, sf, org_config, ensure_records):
+        starting_records = {
+            "Entitlement": [],  # Delete all entitlements so we can delete accounts
+            "Account": [{"Name": "XYZZY"}],
+            "Opportunity": [],  # 0 opportunities
+        }
+        with ensure_records(starting_records):
             with get_org_schema(
-                sf, org_config, include_counts=True, filters=[Filters.populated]
+                sf,
+                org_config,
+                include_counts=True,
+                filters=[Filters.populated],
+                included_objects=["Account", "Contact", "Opportunity"],
             ) as schema:
-                assert "Account" in schema
-                assert "Case" not in schema
+                assert "Account" in schema, schema.keys()
+                assert "Case" not in schema, schema.keys()  # not in included_objects
+                assert (
+                    "Opportunity" not in schema
+                ), schema.keys()  # because not populated
+
+            # Create one case and ensure it is noticed.
+            with ensure_records({"Case": [{}]}):
+                with get_org_schema(
+                    sf, org_config, include_counts=True, filters=[Filters.populated]
+                ) as schema:
+                    assert "Account" in schema, schema.keys()
+                    assert "Case" in schema, schema.keys()
+                    assert (
+                        "Opportunity" not in schema
+                    ), schema.keys()  # because not populated
 
 
 @pytest.mark.needs_org()  # too hard to make these VCR-compatible due to data volume